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 view_options;
24mod window_rewriter;
25
26pub use lookup_table::CreateLookupTableStatement;
27pub use statements::{
28    AlterSourceOperation, CreateSinkStatement, CreateSourceStatement, EmitClause, EmitStrategy,
29    FormatSpec, LateDataClause, ShowCommand, SinkFrom, StreamingStatement, SubscribeStatement,
30    WatermarkDef, WindowFunction,
31};
32pub use view_options::MaterializedViewRetention;
33pub use window_rewriter::WindowRewriter;
34
35use dialect::LaminarDialect;
36use tokenizer::{detect_streaming_ddl, StreamingDdlKind};
37
38/// Parses SQL with streaming extensions.
39///
40/// Routes streaming DDL to custom parsers that use sqlparser's `Parser` API
41/// for structured parsing. Standard SQL is delegated to sqlparser directly.
42///
43/// # Errors
44///
45/// Returns `ParseError` if the SQL syntax is invalid.
46pub fn parse_streaming_sql(sql: &str) -> Result<Vec<StreamingStatement>, ParseError> {
47    StreamingParser::parse_sql(sql).map_err(ParseError::SqlParseError)
48}
49
50/// Parser for streaming SQL extensions.
51///
52/// Provides static methods for parsing streaming SQL statements.
53/// Uses sqlparser's `Parser` API internally for structured parsing
54/// of identifiers, data types, expressions, and queries.
55pub struct StreamingParser;
56
57impl StreamingParser {
58    /// Parse a SQL string with streaming extensions.
59    ///
60    /// Tokenizes the input to detect statement type, then routes to the
61    /// appropriate parser:
62    /// - CREATE SOURCE → `source_parser`
63    /// - CREATE SINK → `sink_parser`
64    /// - CREATE CONTINUOUS QUERY → `continuous_query_parser`
65    /// - Everything else → `sqlparser::parser::Parser`
66    ///
67    /// # Errors
68    ///
69    /// Returns `ParserError` if the SQL syntax is invalid.
70    pub fn parse_sql(sql: &str) -> Result<Vec<StreamingStatement>, sqlparser::parser::ParserError> {
71        let sql_trimmed = sql.trim();
72        if sql_trimmed.is_empty() {
73            return Err(sqlparser::parser::ParserError::ParserError(
74                "Empty SQL statement".to_string(),
75            ));
76        }
77
78        let dialect = LaminarDialect::default();
79
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        let kind = detect_streaming_ddl(&tokens);
86        if kind == StreamingDdlKind::None {
87            return parse_standard_or_temporal_sql(&dialect, sql_trimmed, &tokens);
88        }
89
90        let mut parser =
91            sqlparser::parser::Parser::new(&dialect).with_tokens_with_locations(tokens);
92        let statement = match kind {
93            StreamingDdlKind::CreateSource { .. } => StreamingStatement::CreateSource(Box::new(
94                source_parser::parse_create_source(&mut parser)
95                    .map_err(parse_error_to_parser_error)?,
96            )),
97            StreamingDdlKind::CreateSink { .. } => StreamingStatement::CreateSink(Box::new(
98                sink_parser::parse_create_sink(&mut parser).map_err(parse_error_to_parser_error)?,
99            )),
100            StreamingDdlKind::CreateContinuousQuery { .. } => {
101                continuous_query_parser::parse_continuous_query(&mut parser)
102                    .map_err(parse_error_to_parser_error)?
103            }
104            StreamingDdlKind::DropSource { .. } => {
105                parse_drop_source(&mut parser).map_err(parse_error_to_parser_error)?
106            }
107            StreamingDdlKind::DropSink { .. } => {
108                parse_drop_sink(&mut parser).map_err(parse_error_to_parser_error)?
109            }
110            StreamingDdlKind::DropMaterializedView { .. } => {
111                parse_drop_materialized_view(&mut parser).map_err(parse_error_to_parser_error)?
112            }
113            StreamingDdlKind::ShowSources => StreamingStatement::Show(ShowCommand::Sources),
114            StreamingDdlKind::ShowSinks => StreamingStatement::Show(ShowCommand::Sinks),
115            StreamingDdlKind::ShowQueries => StreamingStatement::Show(ShowCommand::Queries),
116            StreamingDdlKind::ShowMaterializedViews => {
117                StreamingStatement::Show(ShowCommand::MaterializedViews)
118            }
119            StreamingDdlKind::DescribeSource => {
120                parse_describe(&mut parser).map_err(parse_error_to_parser_error)?
121            }
122            StreamingDdlKind::ExplainStreaming => {
123                parse_explain(&mut parser, sql_trimmed).map_err(parse_error_to_parser_error)?
124            }
125            StreamingDdlKind::CreateMaterializedView { .. } => {
126                parse_create_materialized_view(&mut parser, sql_trimmed)
127                    .map_err(parse_error_to_parser_error)?
128            }
129            StreamingDdlKind::CreateStream { .. } => parse_create_stream(&mut parser, sql_trimmed)
130                .map_err(parse_error_to_parser_error)?,
131            StreamingDdlKind::DropStream { .. } => {
132                parse_drop_stream(&mut parser).map_err(parse_error_to_parser_error)?
133            }
134            StreamingDdlKind::ShowStreams => StreamingStatement::Show(ShowCommand::Streams),
135            StreamingDdlKind::ShowTables => StreamingStatement::Show(ShowCommand::Tables),
136            StreamingDdlKind::CreateLookupTable { .. } => {
137                StreamingStatement::CreateLookupTable(Box::new(
138                    lookup_table::parse_create_lookup_table(&mut parser)
139                        .map_err(parse_error_to_parser_error)?,
140                ))
141            }
142            StreamingDdlKind::DropLookupTable { .. } => {
143                let (name, if_exists) = lookup_table::parse_drop_lookup_table(&mut parser)
144                    .map_err(parse_error_to_parser_error)?;
145                StreamingStatement::DropLookupTable { name, if_exists }
146            }
147            StreamingDdlKind::AlterSource => {
148                parse_alter_source(&mut parser).map_err(parse_error_to_parser_error)?
149            }
150            StreamingDdlKind::ShowCheckpointStatus => {
151                StreamingStatement::Show(ShowCommand::CheckpointStatus)
152            }
153            StreamingDdlKind::ShowCreateSource => {
154                parse_show_create_source(&mut parser).map_err(parse_error_to_parser_error)?
155            }
156            StreamingDdlKind::ShowCreateSink => {
157                parse_show_create_sink(&mut parser).map_err(parse_error_to_parser_error)?
158            }
159            StreamingDdlKind::Checkpoint => StreamingStatement::Checkpoint,
160            StreamingDdlKind::Subscribe => StreamingStatement::Subscribe(Box::new(
161                subscribe_parser::parse_subscribe(&mut parser)
162                    .map_err(parse_error_to_parser_error)?,
163            )),
164            StreamingDdlKind::DeclareCursor => declare_parser::parse_declare_cursor(&mut parser)
165                .map_err(parse_error_to_parser_error)?,
166            StreamingDdlKind::RestoreCheckpoint => {
167                parse_restore_checkpoint(&mut parser).map_err(parse_error_to_parser_error)?
168            }
169            StreamingDdlKind::None => unreachable!("handled before parser construction"),
170        };
171        Ok(vec![statement])
172    }
173
174    /// Check if an expression contains a window function.
175    #[must_use]
176    pub fn has_window_function(expr: &sqlparser::ast::Expr) -> bool {
177        match expr {
178            sqlparser::ast::Expr::Function(func) => {
179                if let Some(name) = func.name.0.last() {
180                    let func_name = name.to_string().to_uppercase();
181                    matches!(func_name.as_str(), "TUMBLE" | "HOP" | "SESSION")
182                } else {
183                    false
184                }
185            }
186            _ => false,
187        }
188    }
189
190    /// Parse EMIT clause from SQL string.
191    ///
192    /// # Errors
193    ///
194    /// Returns `ParseError::StreamingError` if the EMIT clause syntax is invalid.
195    pub fn parse_emit_clause(sql: &str) -> Result<Option<EmitClause>, ParseError> {
196        emit_parser::parse_emit_clause_from_sql(sql)
197    }
198
199    /// Parse late data handling clause from SQL string.
200    ///
201    /// # Errors
202    ///
203    /// Returns `ParseError::StreamingError` if the clause syntax is invalid.
204    pub fn parse_late_data_clause(sql: &str) -> Result<Option<LateDataClause>, ParseError> {
205        late_data_parser::parse_late_data_clause_from_sql(sql)
206    }
207}
208
209fn parse_standard_or_temporal_sql(
210    dialect: &LaminarDialect,
211    sql: &str,
212    tokens: &[sqlparser::tokenizer::TokenWithSpan],
213) -> Result<Vec<StreamingStatement>, sqlparser::parser::ParserError> {
214    if let Some(parsed) =
215        join_parser::parse_temporal_probe_query(tokens).map_err(parse_error_to_parser_error)?
216    {
217        return Ok(vec![StreamingStatement::TemporalProbeQuery {
218            statement: Box::new(parsed.statement),
219            analysis: Box::new(parsed.analysis),
220        }]);
221    }
222
223    let statements = sqlparser::parser::Parser::parse_sql(dialect, sql)?;
224    Ok(statements
225        .into_iter()
226        .map(convert_standard_statement)
227        .collect())
228}
229
230/// Convert `ParseError` to `ParserError` for backward compatibility.
231fn parse_error_to_parser_error(e: ParseError) -> sqlparser::parser::ParserError {
232    match e {
233        ParseError::SqlParseError(pe) => pe,
234        ParseError::StreamingError(msg) => sqlparser::parser::ParserError::ParserError(msg),
235        ParseError::WindowError(msg) => {
236            sqlparser::parser::ParserError::ParserError(format!("Window error: {msg}"))
237        }
238        ParseError::ValidationError(msg) => {
239            sqlparser::parser::ParserError::ParserError(format!("Validation error: {msg}"))
240        }
241    }
242}
243
244/// Parse a RESTORE FROM CHECKPOINT statement.
245///
246/// Syntax: `RESTORE FROM CHECKPOINT <id>`
247///
248/// # Errors
249///
250/// Returns `ParseError` if the statement syntax is invalid.
251fn parse_restore_checkpoint(
252    parser: &mut sqlparser::parser::Parser,
253) -> Result<StreamingStatement, ParseError> {
254    // Consume RESTORE
255    tokenizer::expect_custom_keyword(parser, "RESTORE")?;
256    // Consume FROM
257    if !parser.parse_keyword(sqlparser::keywords::Keyword::FROM) {
258        return Err(ParseError::StreamingError(
259            "Expected FROM after RESTORE".to_string(),
260        ));
261    }
262    // Consume CHECKPOINT
263    tokenizer::expect_custom_keyword(parser, "CHECKPOINT")?;
264    // Parse checkpoint ID (numeric literal)
265    let token = parser.next_token();
266    match &token.token {
267        sqlparser::tokenizer::Token::Number(n, _) => {
268            let id: u64 = n
269                .parse()
270                .map_err(|_| ParseError::StreamingError(format!("Invalid checkpoint ID: {n}")))?;
271            Ok(StreamingStatement::RestoreCheckpoint { checkpoint_id: id })
272        }
273        other => Err(ParseError::StreamingError(format!(
274            "Expected checkpoint ID (number), found {other}"
275        ))),
276    }
277}
278
279/// Convert a standard sqlparser statement to a `StreamingStatement`.
280///
281/// Detects INSERT INTO statements and converts them to the streaming
282/// `InsertInto` variant. All other statements are wrapped as `Standard`.
283fn convert_standard_statement(stmt: sqlparser::ast::Statement) -> StreamingStatement {
284    if let sqlparser::ast::Statement::Insert(insert) = &stmt {
285        // Extract table name from TableObject
286        if let sqlparser::ast::TableObject::TableName(ref name) = insert.table {
287            let table_name = name.clone();
288            let columns = insert.columns.clone();
289
290            // Try to extract VALUES rows from source query
291            if let Some(ref source) = insert.source {
292                if let sqlparser::ast::SetExpr::Values(ref values) = *source.body {
293                    let rows = values.rows.clone();
294                    return StreamingStatement::InsertInto {
295                        table_name,
296                        columns,
297                        values: rows,
298                    };
299                }
300            }
301        }
302    }
303    StreamingStatement::Standard(Box::new(stmt))
304}
305
306/// Parse a DROP SOURCE statement.
307///
308/// Syntax: `DROP SOURCE [IF EXISTS] name [CASCADE]`
309///
310/// # Errors
311///
312/// Returns `ParseError` if the statement syntax is invalid.
313fn parse_drop_source(
314    parser: &mut sqlparser::parser::Parser,
315) -> Result<StreamingStatement, ParseError> {
316    parser
317        .expect_keyword(sqlparser::keywords::Keyword::DROP)
318        .map_err(ParseError::SqlParseError)?;
319    tokenizer::expect_custom_keyword(parser, "SOURCE")?;
320    let if_exists = parser.parse_keywords(&[
321        sqlparser::keywords::Keyword::IF,
322        sqlparser::keywords::Keyword::EXISTS,
323    ]);
324    let name = parser
325        .parse_object_name(false)
326        .map_err(ParseError::SqlParseError)?;
327    let cascade = parser.parse_keyword(sqlparser::keywords::Keyword::CASCADE);
328    Ok(StreamingStatement::DropSource {
329        name,
330        if_exists,
331        cascade,
332    })
333}
334
335/// Parse a DROP SINK statement.
336///
337/// Syntax: `DROP SINK [IF EXISTS] name [CASCADE]`
338///
339/// # Errors
340///
341/// Returns `ParseError` if the statement syntax is invalid.
342fn parse_drop_sink(
343    parser: &mut sqlparser::parser::Parser,
344) -> Result<StreamingStatement, ParseError> {
345    parser
346        .expect_keyword(sqlparser::keywords::Keyword::DROP)
347        .map_err(ParseError::SqlParseError)?;
348    tokenizer::expect_custom_keyword(parser, "SINK")?;
349    let if_exists = parser.parse_keywords(&[
350        sqlparser::keywords::Keyword::IF,
351        sqlparser::keywords::Keyword::EXISTS,
352    ]);
353    let name = parser
354        .parse_object_name(false)
355        .map_err(ParseError::SqlParseError)?;
356    let cascade = parser.parse_keyword(sqlparser::keywords::Keyword::CASCADE);
357    Ok(StreamingStatement::DropSink {
358        name,
359        if_exists,
360        cascade,
361    })
362}
363
364/// Parse a DROP MATERIALIZED VIEW statement.
365///
366/// Syntax: `DROP MATERIALIZED VIEW [IF EXISTS] name [CASCADE]`
367///
368/// # Errors
369///
370/// Returns `ParseError` if the statement syntax is invalid.
371fn parse_drop_materialized_view(
372    parser: &mut sqlparser::parser::Parser,
373) -> Result<StreamingStatement, ParseError> {
374    parser
375        .expect_keyword(sqlparser::keywords::Keyword::DROP)
376        .map_err(ParseError::SqlParseError)?;
377    parser
378        .expect_keyword(sqlparser::keywords::Keyword::MATERIALIZED)
379        .map_err(ParseError::SqlParseError)?;
380    parser
381        .expect_keyword(sqlparser::keywords::Keyword::VIEW)
382        .map_err(ParseError::SqlParseError)?;
383    let if_exists = parser.parse_keywords(&[
384        sqlparser::keywords::Keyword::IF,
385        sqlparser::keywords::Keyword::EXISTS,
386    ]);
387    let name = parser
388        .parse_object_name(false)
389        .map_err(ParseError::SqlParseError)?;
390    let cascade = parser.parse_keyword(sqlparser::keywords::Keyword::CASCADE);
391    Ok(StreamingStatement::DropMaterializedView {
392        name,
393        if_exists,
394        cascade,
395    })
396}
397
398/// Parse a CREATE STREAM statement.
399///
400/// Syntax: `CREATE [OR REPLACE] STREAM [IF NOT EXISTS] name AS <select_query> [EMIT <strategy>]`
401///
402/// # Errors
403///
404/// Returns `ParseError` if the statement syntax is invalid.
405fn parse_create_stream(
406    parser: &mut sqlparser::parser::Parser,
407    original_sql: &str,
408) -> Result<StreamingStatement, ParseError> {
409    parser
410        .expect_keyword(sqlparser::keywords::Keyword::CREATE)
411        .map_err(ParseError::SqlParseError)?;
412
413    let or_replace = parser.parse_keywords(&[
414        sqlparser::keywords::Keyword::OR,
415        sqlparser::keywords::Keyword::REPLACE,
416    ]);
417
418    tokenizer::expect_custom_keyword(parser, "STREAM")?;
419
420    let if_not_exists = parser.parse_keywords(&[
421        sqlparser::keywords::Keyword::IF,
422        sqlparser::keywords::Keyword::NOT,
423        sqlparser::keywords::Keyword::EXISTS,
424    ]);
425
426    let name = parser
427        .parse_object_name(false)
428        .map_err(ParseError::SqlParseError)?;
429
430    parser
431        .expect_keyword(sqlparser::keywords::Keyword::AS)
432        .map_err(ParseError::SqlParseError)?;
433
434    // Collect remaining tokens, then peel off the optional trailing
435    // `WITH (...)` before splitting query / EMIT.
436    let remaining = collect_remaining_tokens(parser);
437    let (head_tokens, with_tokens) = split_off_trailing_with(&remaining);
438    let (query_tokens, emit_tokens) = split_at_emit(&head_tokens);
439    let raw_query_sql = query_body_sql(
440        original_sql,
441        &query_tokens,
442        &emit_tokens,
443        with_tokens.as_deref(),
444    );
445
446    let stream_dialect = LaminarDialect::default();
447
448    let (query_stmt, normalized_temporal_sql) = if query_tokens.is_empty() {
449        return Err(ParseError::StreamingError(
450            "Expected SELECT query after AS".to_string(),
451        ));
452    } else if let Some(parsed) = join_parser::parse_temporal_probe_query(&query_tokens)? {
453        (
454            StreamingStatement::TemporalProbeQuery {
455                statement: Box::new(parsed.statement),
456                analysis: Box::new(parsed.analysis),
457            },
458            Some(parsed.normalized_sql),
459        )
460    } else {
461        let mut query_parser = sqlparser::parser::Parser::new(&stream_dialect)
462            .with_tokens_with_locations(query_tokens);
463        let query = query_parser
464            .parse_query()
465            .map_err(ParseError::SqlParseError)?;
466        (
467            StreamingStatement::Standard(Box::new(sqlparser::ast::Statement::Query(query))),
468            None,
469        )
470    };
471    let query_sql = normalized_temporal_sql.unwrap_or(raw_query_sql);
472
473    let emit_clause = if emit_tokens.is_empty() {
474        None
475    } else {
476        let mut emit_parser =
477            sqlparser::parser::Parser::new(&stream_dialect).with_tokens_with_locations(emit_tokens);
478        emit_parser::parse_emit_clause(&mut emit_parser)?
479    };
480
481    let retention_bytes = match with_tokens {
482        None => None,
483        Some(tokens) => {
484            let mut with_parser =
485                sqlparser::parser::Parser::new(&stream_dialect).with_tokens_with_locations(tokens);
486            let opts = tokenizer::parse_with_options(&mut with_parser)?;
487            extract_retention_bytes(&opts)?
488        }
489    };
490
491    Ok(StreamingStatement::CreateStream {
492        name,
493        query: Box::new(query_stmt),
494        emit_clause,
495        or_replace,
496        if_not_exists,
497        query_sql,
498        retention_bytes,
499    })
500}
501
502/// Split off a trailing `WITH (` at depth 0. CTE-style `WITH ident AS (...)`
503/// is ignored because it's followed by an identifier, not `(`.
504fn split_off_trailing_with(
505    tokens: &[sqlparser::tokenizer::TokenWithSpan],
506) -> (
507    Vec<sqlparser::tokenizer::TokenWithSpan>,
508    Option<Vec<sqlparser::tokenizer::TokenWithSpan>>,
509) {
510    use sqlparser::tokenizer::Token;
511    let mut depth: i32 = 0;
512    let mut last_with_idx: Option<usize> = None;
513    for (i, t) in tokens.iter().enumerate() {
514        match &t.token {
515            Token::LParen => depth += 1,
516            Token::RParen => depth -= 1,
517            Token::Word(w) if depth == 0 && w.value.eq_ignore_ascii_case("WITH") => {
518                if matches!(tokens.get(i + 1).map(|t| &t.token), Some(Token::LParen)) {
519                    last_with_idx = Some(i);
520                }
521            }
522            _ => {}
523        }
524    }
525    match last_with_idx {
526        None => (tokens.to_vec(), None),
527        Some(i) => {
528            let mut head = tokens[..i].to_vec();
529            head.push(sqlparser::tokenizer::TokenWithSpan {
530                token: Token::EOF,
531                span: sqlparser::tokenizer::Span::empty(),
532            });
533            let tail = tokens[i..].to_vec();
534            (head, Some(tail))
535        }
536    }
537}
538
539/// Unknown keys are rejected so typos surface immediately.
540fn extract_retention_bytes(
541    opts: &std::collections::HashMap<String, String>,
542) -> Result<Option<u64>, ParseError> {
543    for key in opts.keys() {
544        if !key.eq_ignore_ascii_case("retain_history") {
545            return Err(ParseError::StreamingError(format!(
546                "unknown CREATE STREAM option '{key}' (expected RETAIN_HISTORY)"
547            )));
548        }
549    }
550    match opts
551        .iter()
552        .find(|(k, _)| k.eq_ignore_ascii_case("retain_history"))
553    {
554        None => Ok(None),
555        Some((_, v)) => {
556            let bytes = lookup_table::ByteSize::parse(v)?;
557            Ok(Some(bytes.as_bytes()))
558        }
559    }
560}
561
562/// Slice `original_sql` from the first query token to whichever trailing
563/// clause comes next (`EMIT` or `WITH`), or end of input if neither is
564/// present. Preserves custom streaming syntax that sqlparser's AST would
565/// drop. Falls back to joining token text if spans are empty.
566pub(super) fn query_body_sql(
567    original_sql: &str,
568    query_tokens: &[sqlparser::tokenizer::TokenWithSpan],
569    emit_tokens: &[sqlparser::tokenizer::TokenWithSpan],
570    with_tokens: Option<&[sqlparser::tokenizer::TokenWithSpan]>,
571) -> String {
572    use sqlparser::tokenizer::Token;
573
574    let from_spans = || -> Option<String> {
575        let first = query_tokens
576            .iter()
577            .find(|t| !matches!(t.token, Token::EOF))?;
578        let start = location_to_byte_offset(original_sql, first.span.start)?;
579        let trailer_starts = [emit_tokens.first(), with_tokens.and_then(|t| t.first())]
580            .into_iter()
581            .flatten()
582            .filter_map(|t| location_to_byte_offset(original_sql, t.span.start));
583        let end = trailer_starts.min().unwrap_or(original_sql.len());
584        let slice = original_sql.get(start..end)?;
585        Some(
586            slice
587                .trim_end_matches(|c: char| c.is_whitespace() || c == ';')
588                .to_string(),
589        )
590    };
591
592    from_spans().unwrap_or_else(|| {
593        query_tokens
594            .iter()
595            .take_while(|t| !matches!(t.token, Token::EOF))
596            .map(|t| t.token.to_string())
597            .collect::<Vec<_>>()
598            .join(" ")
599    })
600}
601
602/// Parse a DROP STREAM statement.
603///
604/// Syntax: `DROP STREAM [IF EXISTS] name [CASCADE]`
605///
606/// # Errors
607///
608/// Returns `ParseError` if the statement syntax is invalid.
609fn parse_drop_stream(
610    parser: &mut sqlparser::parser::Parser,
611) -> Result<StreamingStatement, ParseError> {
612    parser
613        .expect_keyword(sqlparser::keywords::Keyword::DROP)
614        .map_err(ParseError::SqlParseError)?;
615    tokenizer::expect_custom_keyword(parser, "STREAM")?;
616    let if_exists = parser.parse_keywords(&[
617        sqlparser::keywords::Keyword::IF,
618        sqlparser::keywords::Keyword::EXISTS,
619    ]);
620    let name = parser
621        .parse_object_name(false)
622        .map_err(ParseError::SqlParseError)?;
623    let cascade = parser.parse_keyword(sqlparser::keywords::Keyword::CASCADE);
624    Ok(StreamingStatement::DropStream {
625        name,
626        if_exists,
627        cascade,
628    })
629}
630
631/// Parse a DESCRIBE statement.
632///
633/// Syntax: `DESCRIBE [EXTENDED] name`
634/// Parse an ALTER SOURCE statement.
635///
636/// Syntax:
637/// - `ALTER SOURCE name ADD COLUMN col_name data_type`
638/// - `ALTER SOURCE name SET ('key' = 'value', ...)`
639///
640/// # Errors
641///
642/// Returns `ParseError` if the statement syntax is invalid.
643fn parse_alter_source(
644    parser: &mut sqlparser::parser::Parser,
645) -> Result<StreamingStatement, ParseError> {
646    parser
647        .expect_keyword(sqlparser::keywords::Keyword::ALTER)
648        .map_err(ParseError::SqlParseError)?;
649    tokenizer::expect_custom_keyword(parser, "SOURCE")?;
650    let name = parser
651        .parse_object_name(false)
652        .map_err(ParseError::SqlParseError)?;
653
654    // Determine operation: ADD COLUMN or SET
655    if parser.parse_keywords(&[
656        sqlparser::keywords::Keyword::ADD,
657        sqlparser::keywords::Keyword::COLUMN,
658    ]) {
659        // ALTER SOURCE name ADD COLUMN col_name data_type
660        let col_name = parser
661            .parse_identifier()
662            .map_err(ParseError::SqlParseError)?;
663        let data_type = parser
664            .parse_data_type()
665            .map_err(ParseError::SqlParseError)?;
666        let column_def = sqlparser::ast::ColumnDef {
667            name: col_name,
668            data_type,
669            options: vec![],
670        };
671        Ok(StreamingStatement::AlterSource {
672            name,
673            operation: statements::AlterSourceOperation::AddColumn { column_def },
674        })
675    } else if parser.parse_keyword(sqlparser::keywords::Keyword::SET) {
676        // ALTER SOURCE name SET ('key' = 'value', ...)
677        parser
678            .expect_token(&sqlparser::tokenizer::Token::LParen)
679            .map_err(ParseError::SqlParseError)?;
680        #[allow(clippy::disallowed_types)] // cold path: SQL parsing
681        let mut properties = std::collections::HashMap::new();
682        loop {
683            let key = parser
684                .parse_literal_string()
685                .map_err(ParseError::SqlParseError)?;
686            parser
687                .expect_token(&sqlparser::tokenizer::Token::Eq)
688                .map_err(ParseError::SqlParseError)?;
689            let value = parser
690                .parse_literal_string()
691                .map_err(ParseError::SqlParseError)?;
692            properties.insert(key, value);
693            if !parser.consume_token(&sqlparser::tokenizer::Token::Comma) {
694                break;
695            }
696        }
697        parser
698            .expect_token(&sqlparser::tokenizer::Token::RParen)
699            .map_err(ParseError::SqlParseError)?;
700        Ok(StreamingStatement::AlterSource {
701            name,
702            operation: statements::AlterSourceOperation::SetProperties { properties },
703        })
704    } else {
705        Err(ParseError::StreamingError(
706            "Expected ADD COLUMN or SET after ALTER SOURCE <name>".to_string(),
707        ))
708    }
709}
710
711/// Parse a DESCRIBE statement.
712///
713/// # Errors
714///
715/// Returns `ParseError` if the statement syntax is invalid.
716fn parse_describe(
717    parser: &mut sqlparser::parser::Parser,
718) -> Result<StreamingStatement, ParseError> {
719    // Consume DESCRIBE or DESC
720    let token = parser.next_token();
721    match &token.token {
722        sqlparser::tokenizer::Token::Word(w)
723            if w.keyword == sqlparser::keywords::Keyword::DESCRIBE
724                || w.keyword == sqlparser::keywords::Keyword::DESC => {}
725        _ => {
726            return Err(ParseError::StreamingError(
727                "Expected DESCRIBE or DESC".to_string(),
728            ));
729        }
730    }
731    let extended = tokenizer::try_parse_custom_keyword(parser, "EXTENDED");
732    let name = parser
733        .parse_object_name(false)
734        .map_err(ParseError::SqlParseError)?;
735    Ok(StreamingStatement::Describe { name, extended })
736}
737
738/// Parse `SHOW CREATE SOURCE <name>`.
739fn parse_show_create_source(
740    parser: &mut sqlparser::parser::Parser,
741) -> Result<StreamingStatement, ParseError> {
742    // Consume SHOW CREATE SOURCE
743    parser
744        .expect_keyword(sqlparser::keywords::Keyword::SHOW)
745        .map_err(ParseError::SqlParseError)?;
746    parser
747        .expect_keyword(sqlparser::keywords::Keyword::CREATE)
748        .map_err(ParseError::SqlParseError)?;
749    tokenizer::expect_custom_keyword(parser, "SOURCE")?;
750    let name = parser
751        .parse_object_name(false)
752        .map_err(ParseError::SqlParseError)?;
753    Ok(StreamingStatement::Show(ShowCommand::CreateSource { name }))
754}
755
756/// Parse `SHOW CREATE SINK <name>`.
757fn parse_show_create_sink(
758    parser: &mut sqlparser::parser::Parser,
759) -> Result<StreamingStatement, ParseError> {
760    // Consume SHOW CREATE SINK
761    parser
762        .expect_keyword(sqlparser::keywords::Keyword::SHOW)
763        .map_err(ParseError::SqlParseError)?;
764    parser
765        .expect_keyword(sqlparser::keywords::Keyword::CREATE)
766        .map_err(ParseError::SqlParseError)?;
767    tokenizer::expect_custom_keyword(parser, "SINK")?;
768    let name = parser
769        .parse_object_name(false)
770        .map_err(ParseError::SqlParseError)?;
771    Ok(StreamingStatement::Show(ShowCommand::CreateSink { name }))
772}
773
774/// Parse an EXPLAIN [ANALYZE] statement wrapping a streaming query.
775///
776/// Syntax: `EXPLAIN [ANALYZE] <streaming_statement>`
777///
778/// # Errors
779///
780/// Returns `ParseError` if the statement syntax is invalid.
781fn parse_explain(
782    parser: &mut sqlparser::parser::Parser,
783    original_sql: &str,
784) -> Result<StreamingStatement, ParseError> {
785    parser
786        .expect_keyword(sqlparser::keywords::Keyword::EXPLAIN)
787        .map_err(ParseError::SqlParseError)?;
788
789    // Check for optional ANALYZE keyword
790    let analyze = tokenizer::try_parse_custom_keyword(parser, "ANALYZE");
791
792    // Find the position after EXPLAIN [ANALYZE] in the original SQL
793    let explain_prefix_upper = original_sql.to_uppercase();
794    let skip_keyword = if analyze { "ANALYZE" } else { "EXPLAIN" };
795    let inner_start = if analyze {
796        explain_prefix_upper
797            .find("ANALYZE")
798            .map_or(0, |pos| pos + "ANALYZE".len())
799    } else {
800        explain_prefix_upper
801            .find("EXPLAIN")
802            .map_or(0, |pos| pos + "EXPLAIN".len())
803    };
804    let inner_sql = original_sql[inner_start..].trim();
805    let _ = skip_keyword; // suppress unused warning
806
807    // Parse the inner statement recursively
808    let inner_stmts = StreamingParser::parse_sql(inner_sql)?;
809    let inner = inner_stmts.into_iter().next().ok_or_else(|| {
810        sqlparser::parser::ParserError::ParserError("Expected statement after EXPLAIN".to_string())
811    })?;
812    Ok(StreamingStatement::Explain {
813        statement: Box::new(inner),
814        analyze,
815    })
816}
817
818/// Parse a CREATE MATERIALIZED VIEW statement.
819///
820/// Syntax:
821/// ```sql
822/// CREATE [OR REPLACE] MATERIALIZED VIEW [IF NOT EXISTS] name
823/// AS <select_query>
824/// [EMIT <strategy>]
825/// ```
826///
827/// # Errors
828///
829/// Returns `ParseError` if the statement syntax is invalid.
830fn parse_create_materialized_view(
831    parser: &mut sqlparser::parser::Parser,
832    original_sql: &str,
833) -> Result<StreamingStatement, ParseError> {
834    parser
835        .expect_keyword(sqlparser::keywords::Keyword::CREATE)
836        .map_err(ParseError::SqlParseError)?;
837
838    let or_replace = parser.parse_keywords(&[
839        sqlparser::keywords::Keyword::OR,
840        sqlparser::keywords::Keyword::REPLACE,
841    ]);
842
843    parser
844        .expect_keyword(sqlparser::keywords::Keyword::MATERIALIZED)
845        .map_err(ParseError::SqlParseError)?;
846    parser
847        .expect_keyword(sqlparser::keywords::Keyword::VIEW)
848        .map_err(ParseError::SqlParseError)?;
849
850    let if_not_exists = parser.parse_keywords(&[
851        sqlparser::keywords::Keyword::IF,
852        sqlparser::keywords::Keyword::NOT,
853        sqlparser::keywords::Keyword::EXISTS,
854    ]);
855
856    let name = parser
857        .parse_object_name(false)
858        .map_err(ParseError::SqlParseError)?;
859
860    parser
861        .expect_keyword(sqlparser::keywords::Keyword::AS)
862        .map_err(ParseError::SqlParseError)?;
863
864    let remaining = collect_remaining_tokens(parser);
865    let (head_tokens, with_tokens) = split_off_trailing_with(&remaining);
866    let (query_tokens, emit_tokens) = split_at_emit(&head_tokens);
867    let raw_query_sql = query_body_sql(
868        original_sql,
869        &query_tokens,
870        &emit_tokens,
871        with_tokens.as_deref(),
872    );
873    let options = with_tokens
874        .map(view_options::parse_view_options)
875        .transpose()?
876        .unwrap_or_default();
877
878    let mv_dialect = LaminarDialect::default();
879
880    let (query_stmt, normalized_temporal_sql) = if query_tokens.is_empty() {
881        return Err(ParseError::StreamingError(
882            "Expected SELECT query after AS".to_string(),
883        ));
884    } else if let Some(parsed) = join_parser::parse_temporal_probe_query(&query_tokens)? {
885        (
886            StreamingStatement::TemporalProbeQuery {
887                statement: Box::new(parsed.statement),
888                analysis: Box::new(parsed.analysis),
889            },
890            Some(parsed.normalized_sql),
891        )
892    } else {
893        let mut query_parser =
894            sqlparser::parser::Parser::new(&mv_dialect).with_tokens_with_locations(query_tokens);
895        let query = query_parser
896            .parse_query()
897            .map_err(ParseError::SqlParseError)?;
898        (
899            StreamingStatement::Standard(Box::new(sqlparser::ast::Statement::Query(query))),
900            None,
901        )
902    };
903    let query_sql = normalized_temporal_sql.unwrap_or(raw_query_sql);
904
905    let emit_clause = if emit_tokens.is_empty() {
906        None
907    } else {
908        let mut emit_parser =
909            sqlparser::parser::Parser::new(&mv_dialect).with_tokens_with_locations(emit_tokens);
910        emit_parser::parse_emit_clause(&mut emit_parser)?
911    };
912
913    Ok(StreamingStatement::CreateMaterializedView {
914        name,
915        query: Box::new(query_stmt),
916        emit_clause,
917        or_replace,
918        if_not_exists,
919        query_sql,
920        retention: options.retention,
921        freshness: options.freshness,
922    })
923}
924
925/// Byte offset in `sql` for a sqlparser `Location` (1-indexed line/column).
926fn location_to_byte_offset(sql: &str, loc: sqlparser::tokenizer::Location) -> Option<usize> {
927    if loc.line == 0 {
928        return None;
929    }
930    let (mut line, mut col) = (1u64, 1u64);
931    for (idx, ch) in sql.char_indices() {
932        if line == loc.line && col == loc.column {
933            return Some(idx);
934        }
935        if ch == '\n' {
936            line += 1;
937            col = 1;
938        } else {
939            col += 1;
940        }
941    }
942    (line == loc.line && col == loc.column).then_some(sql.len())
943}
944
945/// Collect all remaining tokens from the parser into a Vec.
946fn collect_remaining_tokens(
947    parser: &mut sqlparser::parser::Parser,
948) -> Vec<sqlparser::tokenizer::TokenWithSpan> {
949    let mut tokens = Vec::new();
950    loop {
951        let token = parser.next_token();
952        if token.token == sqlparser::tokenizer::Token::EOF {
953            tokens.push(token);
954            break;
955        }
956        tokens.push(token);
957    }
958    tokens
959}
960
961/// Split tokens at the first standalone EMIT keyword (not inside parentheses).
962///
963/// Returns (query_tokens, emit_tokens) where emit_tokens starts with EMIT
964/// (or is empty if no EMIT found).
965fn split_at_emit(
966    tokens: &[sqlparser::tokenizer::TokenWithSpan],
967) -> (
968    Vec<sqlparser::tokenizer::TokenWithSpan>,
969    Vec<sqlparser::tokenizer::TokenWithSpan>,
970) {
971    let mut depth: i32 = 0;
972    for (i, token) in tokens.iter().enumerate() {
973        match &token.token {
974            sqlparser::tokenizer::Token::LParen => depth += 1,
975            sqlparser::tokenizer::Token::RParen => {
976                depth -= 1;
977            }
978            sqlparser::tokenizer::Token::Word(w)
979                if depth == 0 && w.value.eq_ignore_ascii_case("EMIT") =>
980            {
981                let mut query_tokens = tokens[..i].to_vec();
982                query_tokens.push(sqlparser::tokenizer::TokenWithSpan {
983                    token: sqlparser::tokenizer::Token::EOF,
984                    span: sqlparser::tokenizer::Span::empty(),
985                });
986                let emit_tokens = tokens[i..].to_vec();
987                return (query_tokens, emit_tokens);
988            }
989            _ => {}
990        }
991    }
992    (tokens.to_vec(), vec![])
993}
994
995/// SQL parsing errors
996#[derive(Debug, thiserror::Error)]
997pub enum ParseError {
998    /// Standard SQL parse error
999    #[error("SQL parse error: {0}")]
1000    SqlParseError(#[from] sqlparser::parser::ParserError),
1001
1002    /// Streaming extension parse error
1003    #[error("Streaming SQL error: {0}")]
1004    StreamingError(String),
1005
1006    /// Window function error
1007    #[error("Window function error: {0}")]
1008    WindowError(String),
1009
1010    /// Validation error (e.g., invalid option values)
1011    #[error("Validation error: {0}")]
1012    ValidationError(String),
1013}
1014
1015#[cfg(test)]
1016mod tests;
laminar_sql/parser/mod.rs

laminar_sql/parser/
mod.rs
