laminar_sql/parser/
mod.rs

1//! SQL parser with streaming extensions.
2//!
3//! Routes streaming DDL (CREATE SOURCE/SINK/CONTINUOUS QUERY) to custom
4//! parsers that use sqlparser primitives. Routes standard SQL to sqlparser
5//! with `GenericDialect`.
6
7pub mod aggregation_parser;
8pub mod analytic_parser;
9mod continuous_query_parser;
10mod declare_parser;
11pub(crate) mod dialect;
12mod emit_parser;
13pub mod join_parser;
14mod late_data_parser;
15/// Parser for CREATE/DROP LOOKUP TABLE DDL statements
16pub mod lookup_table;
17pub mod order_analyzer;
18mod sink_parser;
19mod source_parser;
20mod statements;
21mod subscribe_parser;
22mod tokenizer;
23mod window_rewriter;
24
25pub use lookup_table::CreateLookupTableStatement;
26pub use statements::{
27    AlterSourceOperation, CreateSinkStatement, CreateSourceStatement, EmitClause, EmitStrategy,
28    FormatSpec, LateDataClause, ShowCommand, SinkFrom, StreamingStatement, SubscribeStatement,
29    WatermarkDef, WindowFunction,
30};
31pub use window_rewriter::WindowRewriter;
32
33use dialect::LaminarDialect;
34use tokenizer::{detect_streaming_ddl, StreamingDdlKind};
35
36/// Parses SQL with streaming extensions.
37///
38/// Routes streaming DDL to custom parsers that use sqlparser's `Parser` API
39/// for structured parsing. Standard SQL is delegated to sqlparser directly.
40///
41/// # Errors
42///
43/// Returns `ParseError` if the SQL syntax is invalid.
44pub fn parse_streaming_sql(sql: &str) -> Result<Vec<StreamingStatement>, ParseError> {
45    StreamingParser::parse_sql(sql).map_err(ParseError::SqlParseError)
46}
47
48/// Parser for streaming SQL extensions.
49///
50/// Provides static methods for parsing streaming SQL statements.
51/// Uses sqlparser's `Parser` API internally for structured parsing
52/// of identifiers, data types, expressions, and queries.
53pub struct StreamingParser;
54
55impl StreamingParser {
56    /// Parse a SQL string with streaming extensions.
57    ///
58    /// Tokenizes the input to detect statement type, then routes to the
59    /// appropriate parser:
60    /// - CREATE SOURCE → `source_parser`
61    /// - CREATE SINK → `sink_parser`
62    /// - CREATE CONTINUOUS QUERY → `continuous_query_parser`
63    /// - Everything else → `sqlparser::parser::Parser`
64    ///
65    /// # Errors
66    ///
67    /// Returns `ParserError` if the SQL syntax is invalid.
68    #[allow(clippy::too_many_lines)]
69    pub fn parse_sql(sql: &str) -> Result<Vec<StreamingStatement>, sqlparser::parser::ParserError> {
70        let sql_trimmed = sql.trim();
71        if sql_trimmed.is_empty() {
72            return Err(sqlparser::parser::ParserError::ParserError(
73                "Empty SQL statement".to_string(),
74            ));
75        }
76
77        let dialect = LaminarDialect::default();
78
79        // Tokenize to detect statement type (with location for better errors)
80        let tokens = sqlparser::tokenizer::Tokenizer::new(&dialect, sql_trimmed)
81            .tokenize_with_location()
82            .map_err(|e| {
83                sqlparser::parser::ParserError::ParserError(format!("Tokenization error: {e}"))
84            })?;
85
86        // Route based on token-level detection
87        match detect_streaming_ddl(&tokens) {
88            StreamingDdlKind::CreateSource { .. } => {
89                let mut parser =
90                    sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
91                let source = source_parser::parse_create_source(&mut parser)
92                    .map_err(parse_error_to_parser_error)?;
93                Ok(vec![StreamingStatement::CreateSource(Box::new(source))])
94            }
95            StreamingDdlKind::CreateSink { .. } => {
96                let mut parser =
97                    sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
98                let sink = sink_parser::parse_create_sink(&mut parser)
99                    .map_err(parse_error_to_parser_error)?;
100                Ok(vec![StreamingStatement::CreateSink(Box::new(sink))])
101            }
102            StreamingDdlKind::CreateContinuousQuery { .. } => {
103                let mut parser =
104                    sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
105                let stmt = continuous_query_parser::parse_continuous_query(&mut parser)
106                    .map_err(parse_error_to_parser_error)?;
107                Ok(vec![stmt])
108            }
109            StreamingDdlKind::DropSource { .. } => {
110                let mut parser =
111                    sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
112                let stmt = parse_drop_source(&mut parser).map_err(parse_error_to_parser_error)?;
113                Ok(vec![stmt])
114            }
115            StreamingDdlKind::DropSink { .. } => {
116                let mut parser =
117                    sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
118                let stmt = parse_drop_sink(&mut parser).map_err(parse_error_to_parser_error)?;
119                Ok(vec![stmt])
120            }
121            StreamingDdlKind::DropMaterializedView { .. } => {
122                let mut parser =
123                    sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
124                let stmt = parse_drop_materialized_view(&mut parser)
125                    .map_err(parse_error_to_parser_error)?;
126                Ok(vec![stmt])
127            }
128            StreamingDdlKind::ShowSources => {
129                Ok(vec![StreamingStatement::Show(ShowCommand::Sources)])
130            }
131            StreamingDdlKind::ShowSinks => Ok(vec![StreamingStatement::Show(ShowCommand::Sinks)]),
132            StreamingDdlKind::ShowQueries => {
133                Ok(vec![StreamingStatement::Show(ShowCommand::Queries)])
134            }
135            StreamingDdlKind::ShowMaterializedViews => Ok(vec![StreamingStatement::Show(
136                ShowCommand::MaterializedViews,
137            )]),
138            StreamingDdlKind::DescribeSource => {
139                let mut parser =
140                    sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
141                let stmt = parse_describe(&mut parser).map_err(parse_error_to_parser_error)?;
142                Ok(vec![stmt])
143            }
144            StreamingDdlKind::ExplainStreaming => {
145                let mut parser =
146                    sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
147                let stmt =
148                    parse_explain(&mut parser, sql_trimmed).map_err(parse_error_to_parser_error)?;
149                Ok(vec![stmt])
150            }
151            StreamingDdlKind::CreateMaterializedView { .. } => {
152                let mut parser =
153                    sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
154                let stmt = parse_create_materialized_view(&mut parser, sql_trimmed)
155                    .map_err(parse_error_to_parser_error)?;
156                Ok(vec![stmt])
157            }
158            StreamingDdlKind::CreateStream { .. } => {
159                let mut parser =
160                    sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
161                let stmt = parse_create_stream(&mut parser, sql_trimmed)
162                    .map_err(parse_error_to_parser_error)?;
163                Ok(vec![stmt])
164            }
165            StreamingDdlKind::DropStream { .. } => {
166                let mut parser =
167                    sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
168                let stmt = parse_drop_stream(&mut parser).map_err(parse_error_to_parser_error)?;
169                Ok(vec![stmt])
170            }
171            StreamingDdlKind::ShowStreams => {
172                Ok(vec![StreamingStatement::Show(ShowCommand::Streams)])
173            }
174            StreamingDdlKind::ShowTables => Ok(vec![StreamingStatement::Show(ShowCommand::Tables)]),
175            StreamingDdlKind::CreateLookupTable { .. } => {
176                let mut parser =
177                    sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
178                let lt = lookup_table::parse_create_lookup_table(&mut parser)
179                    .map_err(parse_error_to_parser_error)?;
180                Ok(vec![StreamingStatement::CreateLookupTable(Box::new(lt))])
181            }
182            StreamingDdlKind::DropLookupTable { .. } => {
183                let mut parser =
184                    sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
185                let (name, if_exists) = lookup_table::parse_drop_lookup_table(&mut parser)
186                    .map_err(parse_error_to_parser_error)?;
187                Ok(vec![StreamingStatement::DropLookupTable {
188                    name,
189                    if_exists,
190                }])
191            }
192            StreamingDdlKind::AlterSource => {
193                let mut parser =
194                    sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
195                let stmt = parse_alter_source(&mut parser).map_err(parse_error_to_parser_error)?;
196                Ok(vec![stmt])
197            }
198            StreamingDdlKind::ShowCheckpointStatus => Ok(vec![StreamingStatement::Show(
199                ShowCommand::CheckpointStatus,
200            )]),
201            StreamingDdlKind::ShowCreateSource => {
202                let mut parser =
203                    sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
204                let stmt =
205                    parse_show_create_source(&mut parser).map_err(parse_error_to_parser_error)?;
206                Ok(vec![stmt])
207            }
208            StreamingDdlKind::ShowCreateSink => {
209                let mut parser =
210                    sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
211                let stmt =
212                    parse_show_create_sink(&mut parser).map_err(parse_error_to_parser_error)?;
213                Ok(vec![stmt])
214            }
215            StreamingDdlKind::Checkpoint => Ok(vec![StreamingStatement::Checkpoint]),
216            StreamingDdlKind::Subscribe => {
217                let mut parser =
218                    sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
219                let stmt = subscribe_parser::parse_subscribe(&mut parser)
220                    .map_err(parse_error_to_parser_error)?;
221                Ok(vec![StreamingStatement::Subscribe(Box::new(stmt))])
222            }
223            StreamingDdlKind::DeclareCursor => {
224                let mut parser =
225                    sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
226                let stmt = declare_parser::parse_declare_cursor(&mut parser)
227                    .map_err(parse_error_to_parser_error)?;
228                Ok(vec![stmt])
229            }
230            StreamingDdlKind::RestoreCheckpoint => {
231                let mut parser =
232                    sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
233                let stmt =
234                    parse_restore_checkpoint(&mut parser).map_err(parse_error_to_parser_error)?;
235                Ok(vec![stmt])
236            }
237            StreamingDdlKind::None => {
238                // Standard SQL - check for INSERT INTO and convert
239                let statements = sqlparser::parser::Parser::parse_sql(&dialect, sql_trimmed)?;
240                Ok(statements
241                    .into_iter()
242                    .map(convert_standard_statement)
243                    .collect())
244            }
245        }
246    }
247
248    /// Check if an expression contains a window function.
249    #[must_use]
250    pub fn has_window_function(expr: &sqlparser::ast::Expr) -> bool {
251        match expr {
252            sqlparser::ast::Expr::Function(func) => {
253                if let Some(name) = func.name.0.last() {
254                    let func_name = name.to_string().to_uppercase();
255                    matches!(func_name.as_str(), "TUMBLE" | "HOP" | "SESSION")
256                } else {
257                    false
258                }
259            }
260            _ => false,
261        }
262    }
263
264    /// Parse EMIT clause from SQL string.
265    ///
266    /// # Errors
267    ///
268    /// Returns `ParseError::StreamingError` if the EMIT clause syntax is invalid.
269    pub fn parse_emit_clause(sql: &str) -> Result<Option<EmitClause>, ParseError> {
270        emit_parser::parse_emit_clause_from_sql(sql)
271    }
272
273    /// Parse late data handling clause from SQL string.
274    ///
275    /// # Errors
276    ///
277    /// Returns `ParseError::StreamingError` if the clause syntax is invalid.
278    pub fn parse_late_data_clause(sql: &str) -> Result<Option<LateDataClause>, ParseError> {
279        late_data_parser::parse_late_data_clause_from_sql(sql)
280    }
281}
282
283/// Convert `ParseError` to `ParserError` for backward compatibility.
284fn parse_error_to_parser_error(e: ParseError) -> sqlparser::parser::ParserError {
285    match e {
286        ParseError::SqlParseError(pe) => pe,
287        ParseError::StreamingError(msg) => sqlparser::parser::ParserError::ParserError(msg),
288        ParseError::WindowError(msg) => {
289            sqlparser::parser::ParserError::ParserError(format!("Window error: {msg}"))
290        }
291        ParseError::ValidationError(msg) => {
292            sqlparser::parser::ParserError::ParserError(format!("Validation error: {msg}"))
293        }
294    }
295}
296
297/// Parse a RESTORE FROM CHECKPOINT statement.
298///
299/// Syntax: `RESTORE FROM CHECKPOINT <id>`
300///
301/// # Errors
302///
303/// Returns `ParseError` if the statement syntax is invalid.
304fn parse_restore_checkpoint(
305    parser: &mut sqlparser::parser::Parser,
306) -> Result<StreamingStatement, ParseError> {
307    // Consume RESTORE
308    tokenizer::expect_custom_keyword(parser, "RESTORE")?;
309    // Consume FROM
310    if !parser.parse_keyword(sqlparser::keywords::Keyword::FROM) {
311        return Err(ParseError::StreamingError(
312            "Expected FROM after RESTORE".to_string(),
313        ));
314    }
315    // Consume CHECKPOINT
316    tokenizer::expect_custom_keyword(parser, "CHECKPOINT")?;
317    // Parse checkpoint ID (numeric literal)
318    let token = parser.next_token();
319    match &token.token {
320        sqlparser::tokenizer::Token::Number(n, _) => {
321            let id: u64 = n
322                .parse()
323                .map_err(|_| ParseError::StreamingError(format!("Invalid checkpoint ID: {n}")))?;
324            Ok(StreamingStatement::RestoreCheckpoint { checkpoint_id: id })
325        }
326        other => Err(ParseError::StreamingError(format!(
327            "Expected checkpoint ID (number), found {other}"
328        ))),
329    }
330}
331
332/// Convert a standard sqlparser statement to a `StreamingStatement`.
333///
334/// Detects INSERT INTO statements and converts them to the streaming
335/// `InsertInto` variant. All other statements are wrapped as `Standard`.
336fn convert_standard_statement(stmt: sqlparser::ast::Statement) -> StreamingStatement {
337    if let sqlparser::ast::Statement::Insert(insert) = &stmt {
338        // Extract table name from TableObject
339        if let sqlparser::ast::TableObject::TableName(ref name) = insert.table {
340            let table_name = name.clone();
341            let columns = insert.columns.clone();
342
343            // Try to extract VALUES rows from source query
344            if let Some(ref source) = insert.source {
345                if let sqlparser::ast::SetExpr::Values(ref values) = *source.body {
346                    let rows: Vec<Vec<sqlparser::ast::Expr>> = values.rows.clone();
347                    return StreamingStatement::InsertInto {
348                        table_name,
349                        columns,
350                        values: rows,
351                    };
352                }
353            }
354        }
355    }
356    StreamingStatement::Standard(Box::new(stmt))
357}
358
359/// Parse a DROP SOURCE statement.
360///
361/// Syntax: `DROP SOURCE [IF EXISTS] name [CASCADE]`
362///
363/// # Errors
364///
365/// Returns `ParseError` if the statement syntax is invalid.
366fn parse_drop_source(
367    parser: &mut sqlparser::parser::Parser,
368) -> Result<StreamingStatement, ParseError> {
369    parser
370        .expect_keyword(sqlparser::keywords::Keyword::DROP)
371        .map_err(ParseError::SqlParseError)?;
372    tokenizer::expect_custom_keyword(parser, "SOURCE")?;
373    let if_exists = parser.parse_keywords(&[
374        sqlparser::keywords::Keyword::IF,
375        sqlparser::keywords::Keyword::EXISTS,
376    ]);
377    let name = parser
378        .parse_object_name(false)
379        .map_err(ParseError::SqlParseError)?;
380    let cascade = parser.parse_keyword(sqlparser::keywords::Keyword::CASCADE);
381    Ok(StreamingStatement::DropSource {
382        name,
383        if_exists,
384        cascade,
385    })
386}
387
388/// Parse a DROP SINK statement.
389///
390/// Syntax: `DROP SINK [IF EXISTS] name [CASCADE]`
391///
392/// # Errors
393///
394/// Returns `ParseError` if the statement syntax is invalid.
395fn parse_drop_sink(
396    parser: &mut sqlparser::parser::Parser,
397) -> Result<StreamingStatement, ParseError> {
398    parser
399        .expect_keyword(sqlparser::keywords::Keyword::DROP)
400        .map_err(ParseError::SqlParseError)?;
401    tokenizer::expect_custom_keyword(parser, "SINK")?;
402    let if_exists = parser.parse_keywords(&[
403        sqlparser::keywords::Keyword::IF,
404        sqlparser::keywords::Keyword::EXISTS,
405    ]);
406    let name = parser
407        .parse_object_name(false)
408        .map_err(ParseError::SqlParseError)?;
409    let cascade = parser.parse_keyword(sqlparser::keywords::Keyword::CASCADE);
410    Ok(StreamingStatement::DropSink {
411        name,
412        if_exists,
413        cascade,
414    })
415}
416
417/// Parse a DROP MATERIALIZED VIEW statement.
418///
419/// Syntax: `DROP MATERIALIZED VIEW [IF EXISTS] name [CASCADE]`
420///
421/// # Errors
422///
423/// Returns `ParseError` if the statement syntax is invalid.
424fn parse_drop_materialized_view(
425    parser: &mut sqlparser::parser::Parser,
426) -> Result<StreamingStatement, ParseError> {
427    parser
428        .expect_keyword(sqlparser::keywords::Keyword::DROP)
429        .map_err(ParseError::SqlParseError)?;
430    parser
431        .expect_keyword(sqlparser::keywords::Keyword::MATERIALIZED)
432        .map_err(ParseError::SqlParseError)?;
433    parser
434        .expect_keyword(sqlparser::keywords::Keyword::VIEW)
435        .map_err(ParseError::SqlParseError)?;
436    let if_exists = parser.parse_keywords(&[
437        sqlparser::keywords::Keyword::IF,
438        sqlparser::keywords::Keyword::EXISTS,
439    ]);
440    let name = parser
441        .parse_object_name(false)
442        .map_err(ParseError::SqlParseError)?;
443    let cascade = parser.parse_keyword(sqlparser::keywords::Keyword::CASCADE);
444    Ok(StreamingStatement::DropMaterializedView {
445        name,
446        if_exists,
447        cascade,
448    })
449}
450
451/// Parse a CREATE STREAM statement.
452///
453/// Syntax: `CREATE [OR REPLACE] STREAM [IF NOT EXISTS] name AS <select_query> [EMIT <strategy>]`
454///
455/// # Errors
456///
457/// Returns `ParseError` if the statement syntax is invalid.
458fn parse_create_stream(
459    parser: &mut sqlparser::parser::Parser,
460    original_sql: &str,
461) -> Result<StreamingStatement, ParseError> {
462    parser
463        .expect_keyword(sqlparser::keywords::Keyword::CREATE)
464        .map_err(ParseError::SqlParseError)?;
465
466    let or_replace = parser.parse_keywords(&[
467        sqlparser::keywords::Keyword::OR,
468        sqlparser::keywords::Keyword::REPLACE,
469    ]);
470
471    tokenizer::expect_custom_keyword(parser, "STREAM")?;
472
473    let if_not_exists = parser.parse_keywords(&[
474        sqlparser::keywords::Keyword::IF,
475        sqlparser::keywords::Keyword::NOT,
476        sqlparser::keywords::Keyword::EXISTS,
477    ]);
478
479    let name = parser
480        .parse_object_name(false)
481        .map_err(ParseError::SqlParseError)?;
482
483    parser
484        .expect_keyword(sqlparser::keywords::Keyword::AS)
485        .map_err(ParseError::SqlParseError)?;
486
487    // Collect remaining tokens, then peel off the optional trailing
488    // `WITH (...)` before splitting query / EMIT.
489    let remaining = collect_remaining_tokens(parser);
490    let (head_tokens, with_tokens) = split_off_trailing_with(&remaining);
491    let (query_tokens, emit_tokens) = split_at_emit(&head_tokens);
492
493    let query_sql = query_body_sql(
494        original_sql,
495        &query_tokens,
496        &emit_tokens,
497        with_tokens.as_deref(),
498    );
499
500    let stream_dialect = LaminarDialect::default();
501
502    let query = if query_tokens.is_empty() {
503        return Err(ParseError::StreamingError(
504            "Expected SELECT query after AS".to_string(),
505        ));
506    } else {
507        let mut query_parser = sqlparser::parser::Parser::new(&stream_dialect)
508            .with_tokens_with_locations(query_tokens);
509        query_parser
510            .parse_query()
511            .map_err(ParseError::SqlParseError)?
512    };
513
514    let query_stmt =
515        StreamingStatement::Standard(Box::new(sqlparser::ast::Statement::Query(query)));
516
517    let emit_clause = if emit_tokens.is_empty() {
518        None
519    } else {
520        let mut emit_parser =
521            sqlparser::parser::Parser::new(&stream_dialect).with_tokens_with_locations(emit_tokens);
522        emit_parser::parse_emit_clause(&mut emit_parser)?
523    };
524
525    let retention_bytes = match with_tokens {
526        None => None,
527        Some(tokens) => {
528            let mut with_parser =
529                sqlparser::parser::Parser::new(&stream_dialect).with_tokens_with_locations(tokens);
530            let opts = tokenizer::parse_with_options(&mut with_parser)?;
531            extract_retention_bytes(&opts)?
532        }
533    };
534
535    Ok(StreamingStatement::CreateStream {
536        name,
537        query: Box::new(query_stmt),
538        emit_clause,
539        or_replace,
540        if_not_exists,
541        query_sql,
542        retention_bytes,
543    })
544}
545
546/// Split off a trailing `WITH (` at depth 0. CTE-style `WITH ident AS (...)`
547/// is ignored because it's followed by an identifier, not `(`.
548fn split_off_trailing_with(
549    tokens: &[sqlparser::tokenizer::TokenWithSpan],
550) -> (
551    Vec<sqlparser::tokenizer::TokenWithSpan>,
552    Option<Vec<sqlparser::tokenizer::TokenWithSpan>>,
553) {
554    use sqlparser::tokenizer::Token;
555    let mut depth: i32 = 0;
556    let mut last_with_idx: Option<usize> = None;
557    for (i, t) in tokens.iter().enumerate() {
558        match &t.token {
559            Token::LParen => depth += 1,
560            Token::RParen => depth -= 1,
561            Token::Word(w) if depth == 0 && w.value.eq_ignore_ascii_case("WITH") => {
562                if matches!(tokens.get(i + 1).map(|t| &t.token), Some(Token::LParen)) {
563                    last_with_idx = Some(i);
564                }
565            }
566            _ => {}
567        }
568    }
569    match last_with_idx {
570        None => (tokens.to_vec(), None),
571        Some(i) => {
572            let mut head = tokens[..i].to_vec();
573            head.push(sqlparser::tokenizer::TokenWithSpan {
574                token: Token::EOF,
575                span: sqlparser::tokenizer::Span::empty(),
576            });
577            let tail = tokens[i..].to_vec();
578            (head, Some(tail))
579        }
580    }
581}
582
583/// Unknown keys are rejected so typos surface immediately.
584fn extract_retention_bytes(
585    opts: &std::collections::HashMap<String, String>,
586) -> Result<Option<u64>, ParseError> {
587    for key in opts.keys() {
588        if !key.eq_ignore_ascii_case("retain_history") {
589            return Err(ParseError::StreamingError(format!(
590                "unknown CREATE STREAM option '{key}' (expected RETAIN_HISTORY)"
591            )));
592        }
593    }
594    match opts
595        .iter()
596        .find(|(k, _)| k.eq_ignore_ascii_case("retain_history"))
597    {
598        None => Ok(None),
599        Some((_, v)) => {
600            let bytes = lookup_table::ByteSize::parse(v)?;
601            Ok(Some(bytes.as_bytes()))
602        }
603    }
604}
605
606/// Slice `original_sql` from the first query token to whichever trailing
607/// clause comes next (`EMIT` or `WITH`), or end of input if neither is
608/// present. Preserves custom streaming syntax that sqlparser's AST would
609/// drop. Falls back to joining token text if spans are empty.
610pub(super) fn query_body_sql(
611    original_sql: &str,
612    query_tokens: &[sqlparser::tokenizer::TokenWithSpan],
613    emit_tokens: &[sqlparser::tokenizer::TokenWithSpan],
614    with_tokens: Option<&[sqlparser::tokenizer::TokenWithSpan]>,
615) -> String {
616    use sqlparser::tokenizer::Token;
617
618    let from_spans = || -> Option<String> {
619        let first = query_tokens
620            .iter()
621            .find(|t| !matches!(t.token, Token::EOF))?;
622        let start = location_to_byte_offset(original_sql, first.span.start)?;
623        let trailer_starts = [emit_tokens.first(), with_tokens.and_then(|t| t.first())]
624            .into_iter()
625            .flatten()
626            .filter_map(|t| location_to_byte_offset(original_sql, t.span.start));
627        let end = trailer_starts.min().unwrap_or(original_sql.len());
628        let slice = original_sql.get(start..end)?;
629        Some(
630            slice
631                .trim_end_matches(|c: char| c.is_whitespace() || c == ';')
632                .to_string(),
633        )
634    };
635
636    from_spans().unwrap_or_else(|| {
637        query_tokens
638            .iter()
639            .take_while(|t| !matches!(t.token, Token::EOF))
640            .map(|t| t.token.to_string())
641            .collect::<Vec<_>>()
642            .join(" ")
643    })
644}
645
646/// Parse a DROP STREAM statement.
647///
648/// Syntax: `DROP STREAM [IF EXISTS] name [CASCADE]`
649///
650/// # Errors
651///
652/// Returns `ParseError` if the statement syntax is invalid.
653fn parse_drop_stream(
654    parser: &mut sqlparser::parser::Parser,
655) -> Result<StreamingStatement, ParseError> {
656    parser
657        .expect_keyword(sqlparser::keywords::Keyword::DROP)
658        .map_err(ParseError::SqlParseError)?;
659    tokenizer::expect_custom_keyword(parser, "STREAM")?;
660    let if_exists = parser.parse_keywords(&[
661        sqlparser::keywords::Keyword::IF,
662        sqlparser::keywords::Keyword::EXISTS,
663    ]);
664    let name = parser
665        .parse_object_name(false)
666        .map_err(ParseError::SqlParseError)?;
667    let cascade = parser.parse_keyword(sqlparser::keywords::Keyword::CASCADE);
668    Ok(StreamingStatement::DropStream {
669        name,
670        if_exists,
671        cascade,
672    })
673}
674
675/// Parse a DESCRIBE statement.
676///
677/// Syntax: `DESCRIBE [EXTENDED] name`
678/// Parse an ALTER SOURCE statement.
679///
680/// Syntax:
681/// - `ALTER SOURCE name ADD COLUMN col_name data_type`
682/// - `ALTER SOURCE name SET ('key' = 'value', ...)`
683///
684/// # Errors
685///
686/// Returns `ParseError` if the statement syntax is invalid.
687fn parse_alter_source(
688    parser: &mut sqlparser::parser::Parser,
689) -> Result<StreamingStatement, ParseError> {
690    parser
691        .expect_keyword(sqlparser::keywords::Keyword::ALTER)
692        .map_err(ParseError::SqlParseError)?;
693    tokenizer::expect_custom_keyword(parser, "SOURCE")?;
694    let name = parser
695        .parse_object_name(false)
696        .map_err(ParseError::SqlParseError)?;
697
698    // Determine operation: ADD COLUMN or SET
699    if parser.parse_keywords(&[
700        sqlparser::keywords::Keyword::ADD,
701        sqlparser::keywords::Keyword::COLUMN,
702    ]) {
703        // ALTER SOURCE name ADD COLUMN col_name data_type
704        let col_name = parser
705            .parse_identifier()
706            .map_err(ParseError::SqlParseError)?;
707        let data_type = parser
708            .parse_data_type()
709            .map_err(ParseError::SqlParseError)?;
710        let column_def = sqlparser::ast::ColumnDef {
711            name: col_name,
712            data_type,
713            options: vec![],
714        };
715        Ok(StreamingStatement::AlterSource {
716            name,
717            operation: statements::AlterSourceOperation::AddColumn { column_def },
718        })
719    } else if parser.parse_keyword(sqlparser::keywords::Keyword::SET) {
720        // ALTER SOURCE name SET ('key' = 'value', ...)
721        parser
722            .expect_token(&sqlparser::tokenizer::Token::LParen)
723            .map_err(ParseError::SqlParseError)?;
724        #[allow(clippy::disallowed_types)] // cold path: SQL parsing
725        let mut properties = std::collections::HashMap::new();
726        loop {
727            let key = parser
728                .parse_literal_string()
729                .map_err(ParseError::SqlParseError)?;
730            parser
731                .expect_token(&sqlparser::tokenizer::Token::Eq)
732                .map_err(ParseError::SqlParseError)?;
733            let value = parser
734                .parse_literal_string()
735                .map_err(ParseError::SqlParseError)?;
736            properties.insert(key, value);
737            if !parser.consume_token(&sqlparser::tokenizer::Token::Comma) {
738                break;
739            }
740        }
741        parser
742            .expect_token(&sqlparser::tokenizer::Token::RParen)
743            .map_err(ParseError::SqlParseError)?;
744        Ok(StreamingStatement::AlterSource {
745            name,
746            operation: statements::AlterSourceOperation::SetProperties { properties },
747        })
748    } else {
749        Err(ParseError::StreamingError(
750            "Expected ADD COLUMN or SET after ALTER SOURCE <name>".to_string(),
751        ))
752    }
753}
754
755/// Parse a DESCRIBE statement.
756///
757/// # Errors
758///
759/// Returns `ParseError` if the statement syntax is invalid.
760fn parse_describe(
761    parser: &mut sqlparser::parser::Parser,
762) -> Result<StreamingStatement, ParseError> {
763    // Consume DESCRIBE or DESC
764    let token = parser.next_token();
765    match &token.token {
766        sqlparser::tokenizer::Token::Word(w)
767            if w.keyword == sqlparser::keywords::Keyword::DESCRIBE
768                || w.keyword == sqlparser::keywords::Keyword::DESC => {}
769        _ => {
770            return Err(ParseError::StreamingError(
771                "Expected DESCRIBE or DESC".to_string(),
772            ));
773        }
774    }
775    let extended = tokenizer::try_parse_custom_keyword(parser, "EXTENDED");
776    let name = parser
777        .parse_object_name(false)
778        .map_err(ParseError::SqlParseError)?;
779    Ok(StreamingStatement::Describe { name, extended })
780}
781
782/// Parse `SHOW CREATE SOURCE <name>`.
783fn parse_show_create_source(
784    parser: &mut sqlparser::parser::Parser,
785) -> Result<StreamingStatement, ParseError> {
786    // Consume SHOW CREATE SOURCE
787    parser
788        .expect_keyword(sqlparser::keywords::Keyword::SHOW)
789        .map_err(ParseError::SqlParseError)?;
790    parser
791        .expect_keyword(sqlparser::keywords::Keyword::CREATE)
792        .map_err(ParseError::SqlParseError)?;
793    tokenizer::expect_custom_keyword(parser, "SOURCE")?;
794    let name = parser
795        .parse_object_name(false)
796        .map_err(ParseError::SqlParseError)?;
797    Ok(StreamingStatement::Show(ShowCommand::CreateSource { name }))
798}
799
800/// Parse `SHOW CREATE SINK <name>`.
801fn parse_show_create_sink(
802    parser: &mut sqlparser::parser::Parser,
803) -> Result<StreamingStatement, ParseError> {
804    // Consume SHOW CREATE SINK
805    parser
806        .expect_keyword(sqlparser::keywords::Keyword::SHOW)
807        .map_err(ParseError::SqlParseError)?;
808    parser
809        .expect_keyword(sqlparser::keywords::Keyword::CREATE)
810        .map_err(ParseError::SqlParseError)?;
811    tokenizer::expect_custom_keyword(parser, "SINK")?;
812    let name = parser
813        .parse_object_name(false)
814        .map_err(ParseError::SqlParseError)?;
815    Ok(StreamingStatement::Show(ShowCommand::CreateSink { name }))
816}
817
818/// Parse an EXPLAIN [ANALYZE] statement wrapping a streaming query.
819///
820/// Syntax: `EXPLAIN [ANALYZE] <streaming_statement>`
821///
822/// # Errors
823///
824/// Returns `ParseError` if the statement syntax is invalid.
825fn parse_explain(
826    parser: &mut sqlparser::parser::Parser,
827    original_sql: &str,
828) -> Result<StreamingStatement, ParseError> {
829    parser
830        .expect_keyword(sqlparser::keywords::Keyword::EXPLAIN)
831        .map_err(ParseError::SqlParseError)?;
832
833    // Check for optional ANALYZE keyword
834    let analyze = tokenizer::try_parse_custom_keyword(parser, "ANALYZE");
835
836    // Find the position after EXPLAIN [ANALYZE] in the original SQL
837    let explain_prefix_upper = original_sql.to_uppercase();
838    let skip_keyword = if analyze { "ANALYZE" } else { "EXPLAIN" };
839    let inner_start = if analyze {
840        explain_prefix_upper
841            .find("ANALYZE")
842            .map_or(0, |pos| pos + "ANALYZE".len())
843    } else {
844        explain_prefix_upper
845            .find("EXPLAIN")
846            .map_or(0, |pos| pos + "EXPLAIN".len())
847    };
848    let inner_sql = original_sql[inner_start..].trim();
849    let _ = skip_keyword; // suppress unused warning
850
851    // Parse the inner statement recursively
852    let inner_stmts = StreamingParser::parse_sql(inner_sql)?;
853    let inner = inner_stmts.into_iter().next().ok_or_else(|| {
854        sqlparser::parser::ParserError::ParserError("Expected statement after EXPLAIN".to_string())
855    })?;
856    Ok(StreamingStatement::Explain {
857        statement: Box::new(inner),
858        analyze,
859    })
860}
861
862/// Parse a CREATE MATERIALIZED VIEW statement.
863///
864/// Syntax:
865/// ```sql
866/// CREATE [OR REPLACE] MATERIALIZED VIEW [IF NOT EXISTS] name
867/// AS <select_query>
868/// [EMIT <strategy>]
869/// ```
870///
871/// # Errors
872///
873/// Returns `ParseError` if the statement syntax is invalid.
874fn parse_create_materialized_view(
875    parser: &mut sqlparser::parser::Parser,
876    original_sql: &str,
877) -> Result<StreamingStatement, ParseError> {
878    parser
879        .expect_keyword(sqlparser::keywords::Keyword::CREATE)
880        .map_err(ParseError::SqlParseError)?;
881
882    let or_replace = parser.parse_keywords(&[
883        sqlparser::keywords::Keyword::OR,
884        sqlparser::keywords::Keyword::REPLACE,
885    ]);
886
887    parser
888        .expect_keyword(sqlparser::keywords::Keyword::MATERIALIZED)
889        .map_err(ParseError::SqlParseError)?;
890    parser
891        .expect_keyword(sqlparser::keywords::Keyword::VIEW)
892        .map_err(ParseError::SqlParseError)?;
893
894    let if_not_exists = parser.parse_keywords(&[
895        sqlparser::keywords::Keyword::IF,
896        sqlparser::keywords::Keyword::NOT,
897        sqlparser::keywords::Keyword::EXISTS,
898    ]);
899
900    let name = parser
901        .parse_object_name(false)
902        .map_err(ParseError::SqlParseError)?;
903
904    parser
905        .expect_keyword(sqlparser::keywords::Keyword::AS)
906        .map_err(ParseError::SqlParseError)?;
907
908    // Collect remaining tokens and split at EMIT boundary (same strategy as continuous query)
909    let remaining = collect_remaining_tokens(parser);
910    let (query_tokens, emit_tokens) = split_at_emit(&remaining);
911
912    let query_sql = query_body_sql(original_sql, &query_tokens, &emit_tokens, None);
913
914    let mv_dialect = LaminarDialect::default();
915
916    let query = if query_tokens.is_empty() {
917        return Err(ParseError::StreamingError(
918            "Expected SELECT query after AS".to_string(),
919        ));
920    } else {
921        let mut query_parser =
922            sqlparser::parser::Parser::new(&mv_dialect).with_tokens_with_locations(query_tokens);
923        query_parser
924            .parse_query()
925            .map_err(ParseError::SqlParseError)?
926    };
927
928    let query_stmt =
929        StreamingStatement::Standard(Box::new(sqlparser::ast::Statement::Query(query)));
930
931    let emit_clause = if emit_tokens.is_empty() {
932        None
933    } else {
934        let mut emit_parser =
935            sqlparser::parser::Parser::new(&mv_dialect).with_tokens_with_locations(emit_tokens);
936        emit_parser::parse_emit_clause(&mut emit_parser)?
937    };
938
939    Ok(StreamingStatement::CreateMaterializedView {
940        name,
941        query: Box::new(query_stmt),
942        emit_clause,
943        or_replace,
944        if_not_exists,
945        query_sql,
946    })
947}
948
949/// Byte offset in `sql` for a sqlparser `Location` (1-indexed line/column).
950fn location_to_byte_offset(sql: &str, loc: sqlparser::tokenizer::Location) -> Option<usize> {
951    if loc.line == 0 {
952        return None;
953    }
954    let (mut line, mut col) = (1u64, 1u64);
955    for (idx, ch) in sql.char_indices() {
956        if line == loc.line && col == loc.column {
957            return Some(idx);
958        }
959        if ch == '\n' {
960            line += 1;
961            col = 1;
962        } else {
963            col += 1;
964        }
965    }
966    (line == loc.line && col == loc.column).then_some(sql.len())
967}
968
969/// Collect all remaining tokens from the parser into a Vec.
970fn collect_remaining_tokens(
971    parser: &mut sqlparser::parser::Parser,
972) -> Vec<sqlparser::tokenizer::TokenWithSpan> {
973    let mut tokens = Vec::new();
974    loop {
975        let token = parser.next_token();
976        if token.token == sqlparser::tokenizer::Token::EOF {
977            tokens.push(token);
978            break;
979        }
980        tokens.push(token);
981    }
982    tokens
983}
984
985/// Split tokens at the first standalone EMIT keyword (not inside parentheses).
986///
987/// Returns (query_tokens, emit_tokens) where emit_tokens starts with EMIT
988/// (or is empty if no EMIT found).
989fn split_at_emit(
990    tokens: &[sqlparser::tokenizer::TokenWithSpan],
991) -> (
992    Vec<sqlparser::tokenizer::TokenWithSpan>,
993    Vec<sqlparser::tokenizer::TokenWithSpan>,
994) {
995    let mut depth: i32 = 0;
996    for (i, token) in tokens.iter().enumerate() {
997        match &token.token {
998            sqlparser::tokenizer::Token::LParen => depth += 1,
999            sqlparser::tokenizer::Token::RParen => {
1000                depth -= 1;
1001            }
1002            sqlparser::tokenizer::Token::Word(w)
1003                if depth == 0 && w.value.eq_ignore_ascii_case("EMIT") =>
1004            {
1005                let mut query_tokens = tokens[..i].to_vec();
1006                query_tokens.push(sqlparser::tokenizer::TokenWithSpan {
1007                    token: sqlparser::tokenizer::Token::EOF,
1008                    span: sqlparser::tokenizer::Span::empty(),
1009                });
1010                let emit_tokens = tokens[i..].to_vec();
1011                return (query_tokens, emit_tokens);
1012            }
1013            _ => {}
1014        }
1015    }
1016    (tokens.to_vec(), vec![])
1017}
1018
1019/// SQL parsing errors
1020#[derive(Debug, thiserror::Error)]
1021pub enum ParseError {
1022    /// Standard SQL parse error
1023    #[error("SQL parse error: {0}")]
1024    SqlParseError(#[from] sqlparser::parser::ParserError),
1025
1026    /// Streaming extension parse error
1027    #[error("Streaming SQL error: {0}")]
1028    StreamingError(String),
1029
1030    /// Window function error
1031    #[error("Window function error: {0}")]
1032    WindowError(String),
1033
1034    /// Validation error (e.g., invalid option values)
1035    #[error("Validation error: {0}")]
1036    ValidationError(String),
1037}
1038
1039#[cfg(test)]
1040mod tests {
1041    use super::*;
1042
1043    /// Helper to parse SQL and return the first statement.
1044    fn parse_one(sql: &str) -> StreamingStatement {
1045        let stmts = StreamingParser::parse_sql(sql).unwrap();
1046        assert_eq!(stmts.len(), 1, "Expected exactly 1 statement");
1047        stmts.into_iter().next().unwrap()
1048    }
1049
1050    #[test]
1051    fn custom_catalog_parsers_reject_duplicate_options_and_trailing_tokens() {
1052        let duplicates = [
1053            "CREATE SOURCE events (id BIGINT) FROM KAFKA ('topic' = 'a', 'TOPIC' = 'b')",
1054            "CREATE SINK output FROM events INTO KAFKA ('topic' = 'a', 'TOPIC' = 'b')",
1055            "CREATE LOOKUP TABLE users (id BIGINT, PRIMARY KEY (id)) WITH ('connector' = 'postgres', 'CONNECTOR' = 'redis')",
1056        ];
1057        for sql in duplicates {
1058            let error = StreamingParser::parse_sql(sql).unwrap_err().to_string();
1059            assert!(error.contains("duplicate connector option"), "{error}");
1060        }
1061
1062        let trailing = [
1063            "CREATE SOURCE events (id BIGINT) WITH ('connector' = 'kafka') TRAILING",
1064            "CREATE SINK output FROM events WITH ('topic' = 'a') TRAILING",
1065            "CREATE LOOKUP TABLE users (id BIGINT, PRIMARY KEY (id)) WITH ('connector' = 'postgres') TRAILING",
1066        ];
1067        for sql in trailing {
1068            let error = StreamingParser::parse_sql(sql).unwrap_err().to_string();
1069            assert!(error.contains("unexpected trailing token"), "{error}");
1070        }
1071    }
1072
1073    #[test]
1074    fn test_parse_drop_source() {
1075        let stmt = parse_one("DROP SOURCE events");
1076        match stmt {
1077            StreamingStatement::DropSource {
1078                name,
1079                if_exists,
1080                cascade,
1081            } => {
1082                assert_eq!(name.to_string(), "events");
1083                assert!(!if_exists);
1084                assert!(!cascade);
1085            }
1086            _ => panic!("Expected DropSource, got {stmt:?}"),
1087        }
1088    }
1089
1090    #[test]
1091    fn test_parse_drop_source_if_exists() {
1092        let stmt = parse_one("DROP SOURCE IF EXISTS events");
1093        match stmt {
1094            StreamingStatement::DropSource {
1095                name,
1096                if_exists,
1097                cascade,
1098            } => {
1099                assert_eq!(name.to_string(), "events");
1100                assert!(if_exists);
1101                assert!(!cascade);
1102            }
1103            _ => panic!("Expected DropSource, got {stmt:?}"),
1104        }
1105    }
1106
1107    #[test]
1108    fn test_parse_drop_source_cascade() {
1109        let stmt = parse_one("DROP SOURCE IF EXISTS events CASCADE");
1110        match stmt {
1111            StreamingStatement::DropSource {
1112                name,
1113                if_exists,
1114                cascade,
1115            } => {
1116                assert_eq!(name.to_string(), "events");
1117                assert!(if_exists);
1118                assert!(cascade);
1119            }
1120            _ => panic!("Expected DropSource, got {stmt:?}"),
1121        }
1122    }
1123
1124    #[test]
1125    fn test_parse_drop_sink() {
1126        let stmt = parse_one("DROP SINK output");
1127        match stmt {
1128            StreamingStatement::DropSink {
1129                name,
1130                if_exists,
1131                cascade,
1132            } => {
1133                assert_eq!(name.to_string(), "output");
1134                assert!(!if_exists);
1135                assert!(!cascade);
1136            }
1137            _ => panic!("Expected DropSink, got {stmt:?}"),
1138        }
1139    }
1140
1141    #[test]
1142    fn test_parse_drop_sink_if_exists() {
1143        let stmt = parse_one("DROP SINK IF EXISTS output");
1144        match stmt {
1145            StreamingStatement::DropSink {
1146                name,
1147                if_exists,
1148                cascade,
1149            } => {
1150                assert_eq!(name.to_string(), "output");
1151                assert!(if_exists);
1152                assert!(!cascade);
1153            }
1154            _ => panic!("Expected DropSink, got {stmt:?}"),
1155        }
1156    }
1157
1158    #[test]
1159    fn test_parse_drop_sink_cascade() {
1160        let stmt = parse_one("DROP SINK output CASCADE");
1161        match stmt {
1162            StreamingStatement::DropSink {
1163                name,
1164                if_exists,
1165                cascade,
1166            } => {
1167                assert_eq!(name.to_string(), "output");
1168                assert!(!if_exists);
1169                assert!(cascade);
1170            }
1171            _ => panic!("Expected DropSink, got {stmt:?}"),
1172        }
1173    }
1174
1175    #[test]
1176    fn test_parse_drop_materialized_view() {
1177        let stmt = parse_one("DROP MATERIALIZED VIEW live_stats");
1178        match stmt {
1179            StreamingStatement::DropMaterializedView {
1180                name,
1181                if_exists,
1182                cascade,
1183            } => {
1184                assert_eq!(name.to_string(), "live_stats");
1185                assert!(!if_exists);
1186                assert!(!cascade);
1187            }
1188            _ => panic!("Expected DropMaterializedView, got {stmt:?}"),
1189        }
1190    }
1191
1192    #[test]
1193    fn test_parse_drop_materialized_view_if_exists_cascade() {
1194        let stmt = parse_one("DROP MATERIALIZED VIEW IF EXISTS live_stats CASCADE");
1195        match stmt {
1196            StreamingStatement::DropMaterializedView {
1197                name,
1198                if_exists,
1199                cascade,
1200            } => {
1201                assert_eq!(name.to_string(), "live_stats");
1202                assert!(if_exists);
1203                assert!(cascade);
1204            }
1205            _ => panic!("Expected DropMaterializedView, got {stmt:?}"),
1206        }
1207    }
1208
1209    #[test]
1210    fn test_parse_show_sources() {
1211        let stmt = parse_one("SHOW SOURCES");
1212        assert!(matches!(
1213            stmt,
1214            StreamingStatement::Show(ShowCommand::Sources)
1215        ));
1216    }
1217
1218    #[test]
1219    fn test_parse_show_sinks() {
1220        let stmt = parse_one("SHOW SINKS");
1221        assert!(matches!(stmt, StreamingStatement::Show(ShowCommand::Sinks)));
1222    }
1223
1224    #[test]
1225    fn test_parse_show_queries() {
1226        let stmt = parse_one("SHOW QUERIES");
1227        assert!(matches!(
1228            stmt,
1229            StreamingStatement::Show(ShowCommand::Queries)
1230        ));
1231    }
1232
1233    #[test]
1234    fn test_parse_show_materialized_views() {
1235        let stmt = parse_one("SHOW MATERIALIZED VIEWS");
1236        assert!(matches!(
1237            stmt,
1238            StreamingStatement::Show(ShowCommand::MaterializedViews)
1239        ));
1240    }
1241
1242    #[test]
1243    fn test_parse_describe() {
1244        let stmt = parse_one("DESCRIBE events");
1245        match stmt {
1246            StreamingStatement::Describe { name, extended } => {
1247                assert_eq!(name.to_string(), "events");
1248                assert!(!extended);
1249            }
1250            _ => panic!("Expected Describe, got {stmt:?}"),
1251        }
1252    }
1253
1254    #[test]
1255    fn test_parse_describe_extended() {
1256        let stmt = parse_one("DESCRIBE EXTENDED my_schema.events");
1257        match stmt {
1258            StreamingStatement::Describe { name, extended } => {
1259                assert_eq!(name.to_string(), "my_schema.events");
1260                assert!(extended);
1261            }
1262            _ => panic!("Expected Describe, got {stmt:?}"),
1263        }
1264    }
1265
1266    #[test]
1267    fn test_parse_explain_select() {
1268        let stmt = parse_one("EXPLAIN SELECT * FROM events");
1269        match stmt {
1270            StreamingStatement::Explain {
1271                statement, analyze, ..
1272            } => {
1273                assert!(matches!(*statement, StreamingStatement::Standard(_)));
1274                assert!(!analyze);
1275            }
1276            _ => panic!("Expected Explain, got {stmt:?}"),
1277        }
1278    }
1279
1280    #[test]
1281    fn test_parse_explain_create_source() {
1282        let stmt = parse_one("EXPLAIN CREATE SOURCE events (id BIGINT)");
1283        match stmt {
1284            StreamingStatement::Explain { statement, .. } => {
1285                assert!(matches!(*statement, StreamingStatement::CreateSource(_)));
1286            }
1287            _ => panic!("Expected Explain wrapping CreateSource, got {stmt:?}"),
1288        }
1289    }
1290
1291    #[test]
1292    fn test_parse_explain_analyze_select() {
1293        let stmt = parse_one("EXPLAIN ANALYZE SELECT * FROM events");
1294        match stmt {
1295            StreamingStatement::Explain {
1296                statement, analyze, ..
1297            } => {
1298                assert!(matches!(*statement, StreamingStatement::Standard(_)));
1299                assert!(analyze, "Expected analyze=true for EXPLAIN ANALYZE");
1300            }
1301            _ => panic!("Expected Explain, got {stmt:?}"),
1302        }
1303    }
1304
1305    #[test]
1306    fn test_parse_create_materialized_view() {
1307        let stmt = parse_one("CREATE MATERIALIZED VIEW live_stats AS SELECT COUNT(*) FROM events");
1308        match stmt {
1309            StreamingStatement::CreateMaterializedView {
1310                name,
1311                emit_clause,
1312                or_replace,
1313                if_not_exists,
1314                ..
1315            } => {
1316                assert_eq!(name.to_string(), "live_stats");
1317                assert!(emit_clause.is_none());
1318                assert!(!or_replace);
1319                assert!(!if_not_exists);
1320            }
1321            _ => panic!("Expected CreateMaterializedView, got {stmt:?}"),
1322        }
1323    }
1324
1325    #[test]
1326    fn test_parse_create_materialized_view_with_emit() {
1327        let stmt = parse_one(
1328            "CREATE MATERIALIZED VIEW live_stats AS SELECT COUNT(*) FROM events EMIT ON WINDOW CLOSE",
1329        );
1330        match stmt {
1331            StreamingStatement::CreateMaterializedView {
1332                name, emit_clause, ..
1333            } => {
1334                assert_eq!(name.to_string(), "live_stats");
1335                assert_eq!(emit_clause, Some(EmitClause::OnWindowClose));
1336            }
1337            _ => panic!("Expected CreateMaterializedView, got {stmt:?}"),
1338        }
1339    }
1340
1341    #[test]
1342    fn test_parse_create_or_replace_materialized_view() {
1343        let stmt = parse_one(
1344            "CREATE OR REPLACE MATERIALIZED VIEW live_stats AS SELECT COUNT(*) FROM events",
1345        );
1346        match stmt {
1347            StreamingStatement::CreateMaterializedView {
1348                name,
1349                or_replace,
1350                if_not_exists,
1351                ..
1352            } => {
1353                assert_eq!(name.to_string(), "live_stats");
1354                assert!(or_replace);
1355                assert!(!if_not_exists);
1356            }
1357            _ => panic!("Expected CreateMaterializedView, got {stmt:?}"),
1358        }
1359    }
1360
1361    #[test]
1362    fn test_parse_create_materialized_view_if_not_exists() {
1363        let stmt = parse_one(
1364            "CREATE MATERIALIZED VIEW IF NOT EXISTS live_stats AS SELECT COUNT(*) FROM events",
1365        );
1366        match stmt {
1367            StreamingStatement::CreateMaterializedView {
1368                name,
1369                or_replace,
1370                if_not_exists,
1371                ..
1372            } => {
1373                assert_eq!(name.to_string(), "live_stats");
1374                assert!(!or_replace);
1375                assert!(if_not_exists);
1376            }
1377            _ => panic!("Expected CreateMaterializedView, got {stmt:?}"),
1378        }
1379    }
1380
1381    #[test]
1382    fn test_parse_insert_into() {
1383        let stmt = parse_one("INSERT INTO events (id, name) VALUES (1, 'test')");
1384        match stmt {
1385            StreamingStatement::InsertInto {
1386                table_name,
1387                columns,
1388                values,
1389            } => {
1390                assert_eq!(table_name.to_string(), "events");
1391                assert_eq!(columns.len(), 2);
1392                assert_eq!(columns[0].to_string(), "id");
1393                assert_eq!(columns[1].to_string(), "name");
1394                assert_eq!(values.len(), 1);
1395                assert_eq!(values[0].len(), 2);
1396            }
1397            _ => panic!("Expected InsertInto, got {stmt:?}"),
1398        }
1399    }
1400
1401    #[test]
1402    fn test_parse_insert_into_multiple_rows() {
1403        let stmt = parse_one("INSERT INTO events VALUES (1, 'a'), (2, 'b'), (3, 'c')");
1404        match stmt {
1405            StreamingStatement::InsertInto {
1406                table_name,
1407                columns,
1408                values,
1409            } => {
1410                assert_eq!(table_name.to_string(), "events");
1411                assert!(columns.is_empty());
1412                assert_eq!(values.len(), 3);
1413            }
1414            _ => panic!("Expected InsertInto, got {stmt:?}"),
1415        }
1416    }
1417
1418    // ── CREATE STREAM tests ─────────────────────────────
1419
1420    #[test]
1421    fn test_parse_create_stream() {
1422        let stmt = parse_one(
1423            "CREATE STREAM session_activity AS SELECT session_id, COUNT(*) as cnt FROM clicks GROUP BY session_id",
1424        );
1425        match stmt {
1426            StreamingStatement::CreateStream {
1427                name,
1428                or_replace,
1429                if_not_exists,
1430                emit_clause,
1431                ..
1432            } => {
1433                assert_eq!(name.to_string(), "session_activity");
1434                assert!(!or_replace);
1435                assert!(!if_not_exists);
1436                assert!(emit_clause.is_none());
1437            }
1438            _ => panic!("Expected CreateStream, got {stmt:?}"),
1439        }
1440    }
1441
1442    #[test]
1443    fn test_parse_create_or_replace_stream() {
1444        let stmt = parse_one("CREATE OR REPLACE STREAM metrics AS SELECT AVG(value) FROM events");
1445        match stmt {
1446            StreamingStatement::CreateStream { or_replace, .. } => {
1447                assert!(or_replace);
1448            }
1449            _ => panic!("Expected CreateStream, got {stmt:?}"),
1450        }
1451    }
1452
1453    #[test]
1454    fn test_parse_create_stream_if_not_exists() {
1455        let stmt = parse_one("CREATE STREAM IF NOT EXISTS counts AS SELECT COUNT(*) FROM events");
1456        match stmt {
1457            StreamingStatement::CreateStream { if_not_exists, .. } => {
1458                assert!(if_not_exists);
1459            }
1460            _ => panic!("Expected CreateStream, got {stmt:?}"),
1461        }
1462    }
1463
1464    #[test]
1465    fn test_parse_create_stream_with_emit() {
1466        let stmt =
1467            parse_one("CREATE STREAM windowed AS SELECT COUNT(*) FROM events EMIT ON WINDOW CLOSE");
1468        match stmt {
1469            StreamingStatement::CreateStream { emit_clause, .. } => {
1470                assert_eq!(emit_clause, Some(EmitClause::OnWindowClose));
1471            }
1472            _ => panic!("Expected CreateStream, got {stmt:?}"),
1473        }
1474    }
1475
1476    #[test]
1477    fn create_stream_with_retain_history() {
1478        let stmt = parse_one(
1479            "CREATE STREAM trades AS SELECT * FROM events WITH ('retain_history' = '64mb')",
1480        );
1481        let StreamingStatement::CreateStream {
1482            retention_bytes, ..
1483        } = stmt
1484        else {
1485            panic!("expected CreateStream");
1486        };
1487        assert_eq!(retention_bytes, Some(64 * 1024 * 1024));
1488    }
1489
1490    #[test]
1491    fn create_stream_with_retain_history_after_emit() {
1492        let stmt = parse_one(
1493            "CREATE STREAM trades AS SELECT COUNT(*) FROM events EMIT ON WINDOW CLOSE \
1494             WITH ('retain_history' = '8mb')",
1495        );
1496        let StreamingStatement::CreateStream {
1497            retention_bytes,
1498            emit_clause,
1499            ..
1500        } = stmt
1501        else {
1502            panic!("expected CreateStream");
1503        };
1504        assert_eq!(retention_bytes, Some(8 * 1024 * 1024));
1505        assert_eq!(emit_clause, Some(EmitClause::OnWindowClose));
1506    }
1507
1508    #[test]
1509    fn create_stream_rejects_unknown_with_option() {
1510        let res = parse_streaming_sql("CREATE STREAM s AS SELECT * FROM e WITH ('bogus' = '1')");
1511        let err = res.expect_err("unknown WITH key must error");
1512        assert!(err.to_string().contains("bogus"), "got: {err}");
1513    }
1514
1515    #[test]
1516    fn create_stream_with_retain_history_excludes_with_from_query_sql() {
1517        // Regression: query_body_sql used to extend to original_sql.len() when
1518        // EMIT was absent, which sucked the trailing WITH clause into query_sql
1519        // and broke planner re-parsing.
1520        let stmt = parse_one(
1521            "CREATE STREAM trades AS SELECT * FROM events WITH ('retain_history' = '64mb')",
1522        );
1523        let StreamingStatement::CreateStream { query_sql, .. } = stmt else {
1524            panic!("expected CreateStream");
1525        };
1526        assert!(
1527            !query_sql.to_uppercase().contains("RETAIN_HISTORY"),
1528            "query_sql leaked WITH clause: {query_sql}"
1529        );
1530        assert!(
1531            query_sql.contains("FROM events"),
1532            "query_sql should still hold the SELECT body, got: {query_sql}"
1533        );
1534    }
1535
1536    #[test]
1537    fn test_parse_drop_stream() {
1538        let stmt = parse_one("DROP STREAM my_stream");
1539        match stmt {
1540            StreamingStatement::DropStream {
1541                name,
1542                if_exists,
1543                cascade,
1544            } => {
1545                assert_eq!(name.to_string(), "my_stream");
1546                assert!(!if_exists);
1547                assert!(!cascade);
1548            }
1549            _ => panic!("Expected DropStream, got {stmt:?}"),
1550        }
1551    }
1552
1553    #[test]
1554    fn test_parse_drop_stream_if_exists() {
1555        let stmt = parse_one("DROP STREAM IF EXISTS my_stream");
1556        match stmt {
1557            StreamingStatement::DropStream {
1558                name,
1559                if_exists,
1560                cascade,
1561            } => {
1562                assert_eq!(name.to_string(), "my_stream");
1563                assert!(if_exists);
1564                assert!(!cascade);
1565            }
1566            _ => panic!("Expected DropStream, got {stmt:?}"),
1567        }
1568    }
1569
1570    #[test]
1571    fn test_parse_drop_stream_cascade() {
1572        let stmt = parse_one("DROP STREAM my_stream CASCADE");
1573        match stmt {
1574            StreamingStatement::DropStream {
1575                name,
1576                if_exists,
1577                cascade,
1578            } => {
1579                assert_eq!(name.to_string(), "my_stream");
1580                assert!(!if_exists);
1581                assert!(cascade);
1582            }
1583            _ => panic!("Expected DropStream, got {stmt:?}"),
1584        }
1585    }
1586
1587    #[test]
1588    fn test_parse_show_streams() {
1589        let stmt = parse_one("SHOW STREAMS");
1590        assert!(matches!(
1591            stmt,
1592            StreamingStatement::Show(ShowCommand::Streams)
1593        ));
1594    }
1595
1596    #[test]
1597    fn test_parse_alter_source_add_column() {
1598        let stmt = parse_one("ALTER SOURCE events ADD COLUMN new_col INT");
1599        match stmt {
1600            StreamingStatement::AlterSource { name, operation } => {
1601                assert_eq!(name.to_string(), "events");
1602                match operation {
1603                    statements::AlterSourceOperation::AddColumn { column_def } => {
1604                        assert_eq!(column_def.name.value, "new_col");
1605                        assert_eq!(column_def.data_type, sqlparser::ast::DataType::Int(None));
1606                    }
1607                    statements::AlterSourceOperation::SetProperties { .. } => {
1608                        panic!("Expected AddColumn")
1609                    }
1610                }
1611            }
1612            _ => panic!("Expected AlterSource, got {stmt:?}"),
1613        }
1614    }
1615
1616    #[test]
1617    fn test_parse_alter_source_set_properties() {
1618        let stmt = parse_one("ALTER SOURCE events SET ('batch.size' = '1000', 'timeout' = '5s')");
1619        match stmt {
1620            StreamingStatement::AlterSource { name, operation } => {
1621                assert_eq!(name.to_string(), "events");
1622                match operation {
1623                    statements::AlterSourceOperation::SetProperties { properties } => {
1624                        assert_eq!(properties.get("batch.size"), Some(&"1000".to_string()));
1625                        assert_eq!(properties.get("timeout"), Some(&"5s".to_string()));
1626                    }
1627                    statements::AlterSourceOperation::AddColumn { .. } => {
1628                        panic!("Expected SetProperties")
1629                    }
1630                }
1631            }
1632            _ => panic!("Expected AlterSource, got {stmt:?}"),
1633        }
1634    }
1635
1636    #[test]
1637    fn test_parse_checkpoint() {
1638        let stmt = parse_one("CHECKPOINT");
1639        assert!(
1640            matches!(stmt, StreamingStatement::Checkpoint),
1641            "Expected Checkpoint, got {stmt:?}"
1642        );
1643    }
1644
1645    #[test]
1646    fn test_parse_show_checkpoint_status() {
1647        let stmt = parse_one("SHOW CHECKPOINT STATUS");
1648        assert!(
1649            matches!(
1650                stmt,
1651                StreamingStatement::Show(ShowCommand::CheckpointStatus)
1652            ),
1653            "Expected Show(CheckpointStatus), got {stmt:?}"
1654        );
1655    }
1656
1657    #[test]
1658    fn test_parse_restore_checkpoint() {
1659        let stmt = parse_one("RESTORE FROM CHECKPOINT 42");
1660        match stmt {
1661            StreamingStatement::RestoreCheckpoint { checkpoint_id } => {
1662                assert_eq!(checkpoint_id, 42);
1663            }
1664            _ => panic!("Expected RestoreCheckpoint, got {stmt:?}"),
1665        }
1666    }
1667
1668    #[test]
1669    fn test_parse_restore_checkpoint_large_id() {
1670        let stmt = parse_one("RESTORE FROM CHECKPOINT 123456");
1671        match stmt {
1672            StreamingStatement::RestoreCheckpoint { checkpoint_id } => {
1673                assert_eq!(checkpoint_id, 123_456);
1674            }
1675            _ => panic!("Expected RestoreCheckpoint, got {stmt:?}"),
1676        }
1677    }
1678
1679    #[test]
1680    fn test_parse_show_create_source() {
1681        let stmt = parse_one("SHOW CREATE SOURCE events");
1682        match stmt {
1683            StreamingStatement::Show(ShowCommand::CreateSource { name }) => {
1684                assert_eq!(name.to_string(), "events");
1685            }
1686            _ => panic!("Expected Show(CreateSource), got {stmt:?}"),
1687        }
1688    }
1689
1690    #[test]
1691    fn test_parse_show_create_sink() {
1692        let stmt = parse_one("SHOW CREATE SINK output");
1693        match stmt {
1694            StreamingStatement::Show(ShowCommand::CreateSink { name }) => {
1695                assert_eq!(name.to_string(), "output");
1696            }
1697            _ => panic!("Expected Show(CreateSink), got {stmt:?}"),
1698        }
1699    }
1700
1701    #[test]
1702    fn create_stream_preserves_temporal_probe_join() {
1703        let sql = "CREATE STREAM markouts_long AS \
1704                   SELECT t.s, p.offset_ms FROM trade_probe t \
1705                   TEMPORAL PROBE JOIN price_ref r \
1706                       ON (s) TIMESTAMPS (ts, ts) \
1707                       LIST (0s, 1s, 5s, 30s) AS p";
1708        let StreamingStatement::CreateStream { query_sql, .. } = parse_one(sql) else {
1709            panic!("expected CreateStream");
1710        };
1711        assert!(
1712            query_sql.to_uppercase().contains("TEMPORAL PROBE JOIN"),
1713            "got: {query_sql}"
1714        );
1715        assert!(
1716            query_sql.contains("LIST (0s, 1s, 5s, 30s)"),
1717            "got: {query_sql}"
1718        );
1719        assert!(query_sql.contains("AS p"), "got: {query_sql}");
1720    }
1721
1722    #[test]
1723    fn create_stream_temporal_probe_range_step_preserved() {
1724        let sql = "CREATE STREAM mk AS \
1725                   SELECT s FROM trades t TEMPORAL PROBE JOIN prices r \
1726                   ON (symbol) TIMESTAMPS (ts, ts) \
1727                   RANGE FROM 0s TO 30s STEP 5s AS p";
1728        let StreamingStatement::CreateStream { query_sql, .. } = parse_one(sql) else {
1729            panic!("expected CreateStream");
1730        };
1731        assert!(
1732            query_sql.contains("RANGE FROM 0s TO 30s STEP 5s"),
1733            "got: {query_sql}"
1734        );
1735    }
1736
1737    #[test]
1738    fn create_mv_preserves_temporal_probe_join() {
1739        let sql = "CREATE MATERIALIZED VIEW mv AS \
1740                   SELECT t.s FROM trades t TEMPORAL PROBE JOIN prices r \
1741                   ON (s) TIMESTAMPS (ts, ts) LIST (0s, 5s) AS p";
1742        let StreamingStatement::CreateMaterializedView { query_sql, .. } = parse_one(sql) else {
1743            panic!("expected CreateMaterializedView");
1744        };
1745        assert!(
1746            query_sql.to_uppercase().contains("TEMPORAL PROBE JOIN"),
1747            "got: {query_sql}"
1748        );
1749    }
1750
1751    #[test]
1752    fn create_stream_emit_is_not_captured_in_query_sql() {
1753        let sql = "CREATE STREAM s AS SELECT COUNT(*) FROM events EMIT ON WINDOW CLOSE";
1754        let StreamingStatement::CreateStream {
1755            query_sql,
1756            emit_clause,
1757            ..
1758        } = parse_one(sql)
1759        else {
1760            panic!("expected CreateStream");
1761        };
1762        assert_eq!(emit_clause, Some(EmitClause::OnWindowClose));
1763        assert!(
1764            !query_sql.to_uppercase().contains("EMIT"),
1765            "got: {query_sql}"
1766        );
1767        assert!(query_sql.contains("FROM events"), "got: {query_sql}");
1768    }
1769
1770    #[test]
1771    fn create_stream_plain_select_query_sql_executes() {
1772        let sql = "CREATE STREAM counts AS SELECT COUNT(*) AS c FROM events";
1773        let StreamingStatement::CreateStream { query_sql, .. } = parse_one(sql) else {
1774            panic!("expected CreateStream");
1775        };
1776        assert!(query_sql.to_uppercase().contains("SELECT"));
1777        assert!(query_sql.contains("FROM events"));
1778    }
1779}
laminar_sql/parser/mod.rs

laminar_sql/parser/
mod.rs
