diff --git a/CLAUDE.md b/CLAUDE.md index 8367ae54f..35e47637c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -121,6 +121,7 @@ node scripts/compare-corpus-reports.js target/corpus-report.json target/corpus-r - **Refresh baseline after each accepted commit**: `cp target/corpus-report.json target/corpus-report-baseline.json`. Otherwise `compare-corpus-reports.js` credits old deltas and can hide fresh regressions. - **If the symlinked corpus changes mid-session** (someone runs `make process` in kernel-cll-corpus — it can add tens of thousands of files), your saved baseline is stale and deltas are meaningless (you'll see bogus +50k/-16k swings). Re-isolate *your* change: `git stash` your edits → rebuild + run corpus → `cp` baseline → `git stash pop` → rebuild + run + compare. - `compare-corpus-reports.js` only lists *added* tests under "New Tests" — deleted/pruned files don't appear there. After a kernel-cll-corpus pipeline run, also check `git status -s` in that repo to see what was removed. +- **When the corpus drifts mid-session, trust a file-by-file regression check, not `compare-corpus-reports.js` aggregate deltas.** Compare the two reports' `test_results` dicts in Python for keys that went `pass`→`fail`: `reg=[k for k in cur if cur[k].startswith('fail') and base.get(k,'').startswith('pass')]`. Zero `reg` = no regressions even when totals shifted by thousands. - **Pipeline reprocess (`make process` in kernel-cll-corpus) takes ~10 minutes** for the full corpus. Don't poll — arm a Monitor on `while pgrep -f pipeline.process; do sleep 15; done` and let it wake you. - **Anonymizer-corruption signature**: `'s'` (the `'s'` placeholder string directly abutting an identifier/keyword, e.g. `'s'HOUR`, `'s'id_5`) is unique to anonymizer misalignment. Filter on exactly `'s'` — a broader `''` regex misaligns on multi-string SQL (`'foo','bar'`) and silently deletes hand-written sqlglot fixtures. - **Query-log truncation heuristics** (`pipeline/process.py::_looks_truncated`) that worked without false positives: trailing punctuation (`,`/`(`/`=`/operator), trailing clause keyword (SELECT/FROM/BY/AS/…), and `CASE` count > `END` count. Removed ~4k Redshift query-log fragments. @@ -271,6 +272,8 @@ while depth > 0 { ``` Use this pattern for dialect-specific clauses that don't need AST representation. +**Unsupported / opaque statement nodes (prefer over masquerades):** Unmodelled DDL returns honest nodes that preserve the raw body, NOT a fake `Comment`/`SetVariable`: `CreateUnsupported`/`DropUnsupported`/`AlterUnsupported { object_type, body }` and `UnsupportedStatement { keyword, body }` (SYSTEM/LOCK/…). Helpers in `src/parser/mod.rs`: `skip_to_statement_end()` (paren-aware skip to `;`/EOF), `consume_statement_body_text()` (raw text capture — **filters `Token::Whitespace`**, else joined strings get stray spaces), `parse_create_body_statement()` (parse a body as a `Statement`, but **revert + skip if it doesn't fully consume to `;`/EOF** — guards against partial parses like `EXECUTE DBT PROJECT`), and `parse_function_body_definition()` (shared `$$…$$`/`'…'` body capture with correct `body_start`, used by `CREATE PROCEDURE` and `DO`). + **Reserved-keyword-as-alias carve-out** (recurring fix shape): - When a keyword (e.g. CLUSTER, SORT, FINAL, AT, BEFORE) is reserved only because of a specific clause (`CLUSTER BY`, `t AT(...)` time-travel), accept it as an identifier alias in `parse_optional_alias` when the lookahead doesn't match the clause shape (next-not-`BY`, next-not-`(`, etc.). - Existing instances in `parse_optional_alias` for CLUSTER / SORT / FINAL / VIEW / OPTION / USE/IGNORE/FORCE in `parse_table_factor` for AT / BEFORE serve as templates — copy the matching block rather than reinventing. @@ -295,6 +298,8 @@ The tokenizer greedily folds a trailing `.` into the number token, so `proj-NNN. - `ClickHouseDialect`, `SnowflakeDialect`, `BigQueryDialect`, `PostgreSqlDialect` - `MySqlDialect`, `MsSqlDialect`, `SQLiteDialect` +**Dialect-specific AST node names** match the dialect-struct casing: `DuckDb` (not `Duckdb`), `ClickHouse`, `BigQuery`, `MsSql`, `MySql` — e.g. `DuckDbLoad`, `DatabricksMap`, `CopyIntoSnowflake`. Prefix vs suffix is inconsistent across existing nodes; match the casing, pick whichever reads better. + ## Development Guidelines ### Syntax vs Semantics @@ -354,6 +359,8 @@ Since this is a fork of apache/datafusion-sqlparser-rs: - `cargo nextest run` without `RUST_MIN_STACK=8388608` always SIGABRTs three tests (`parse_deeply_nested_expr_hits_recursion_limits`, `..._parens_...`, `..._subquery_...`). These are stack-size probes, not regressions — ignore if the rest of the run is green. - Avoid adding serde dependency to test code - use manual JSON building if needed - `verified_stmt(s)` requires `s` to be a full statement that round-trips (parse → display → equal). For fragment / coverage tests on a non-statement (a bare CAST, a SELECT-list snippet), wrap in `SELECT` or call `parse_sql_statements(input).unwrap()` directly — `verified_stmt` will otherwise fail with `Expected an SQL statement, found: ...`. +- **Spans are dummy (`Location` line/col 0) under `Parser::parse_sql` / `verified_stmt` / the `cli` example** — they call `with_tokens` which installs `Span::default()`. To assert real `WithSpan` / `body_start` locations, build the parser explicitly: `Parser::new(&dialect).with_tokens_with_locations(Tokenizer::new(&dialect, sql).tokenize_with_location().unwrap())`. See `parse_sql_with_locations` in `tests/redshift_customer_procedure_samples.rs`. +- `one_statement_parses_to(sql, canonical)` re-parses `canonical` and asserts **full AST equality**. If your node's `Display` reconstructs to text that re-parses through a *different* code path (e.g. `FOREIGN TABLE` → `CREATE EXTERNAL TABLE`, which sets `hive_formats` differently), it fails on a field mismatch — use `parse_sql_statements` + manual field asserts instead. **Running corpus tests (now fast!):** ```bash diff --git a/src/ast/mod.rs b/src/ast/mod.rs index b067cee0d..b79a77972 100644 --- a/src/ast/mod.rs +++ b/src/ast/mod.rs @@ -1745,9 +1745,12 @@ pub enum Statement { extension_name: WithSpan, }, /// ```sql - /// LOAD + /// LOAD /// ``` - Load { + /// + /// DuckDB extension load. Named `DuckDbLoad` to disambiguate from the + /// Hive/Spark/BigQuery `LOAD DATA ...` statement ([`Statement::LoadData`]). + DuckDbLoad { /// Only for DuckDB extension_name: WithSpan, }, @@ -2657,6 +2660,183 @@ pub enum Statement { body_definition: Option, }, /// ```sql + /// CREATE [ OR REPLACE | OR ALTER ] TASK [ IF NOT EXISTS ] + /// [ ... ] + /// [ AFTER [ , ... ] ] + /// [ WHEN ] + /// AS + /// ``` + /// + /// Snowflake task. The scheduling / warehouse / session properties carry no + /// lineage and are consumed without being represented. The `AS ` body + /// is the statement the task runs and is the primary lineage input; `after` + /// lists predecessor tasks (task-graph edges) and `when` is the gating + /// predicate (often references a stream via `SYSTEM$STREAM_HAS_DATA`). + /// `body` is `None` only when the body could not be parsed (e.g. an opaque + /// `EXECUTE DBT PROJECT`), in which case the remainder was skipped. + /// See: + CreateTask { + or_replace: bool, + or_alter: bool, + if_not_exists: bool, + name: WithSpan, + after: Vec>, + when: Option>, + body: Option>, + }, + /// ```sql + /// CREATE [ OR REPLACE ] STREAM [ IF NOT EXISTS ] + /// [ COPY GRANTS ] ON [ { AT | BEFORE } ( ... ) ] + /// [ ... ] + /// ``` + /// + /// Snowflake change-data-capture stream. A stream tracks DML on a source + /// object, so the lineage edge is `source -> stream`. `source_type` is the + /// object kind the stream is defined on (`TABLE`, `VIEW`, `STAGE`, + /// `EXTERNAL TABLE`, ...) and `source` is its name. Time-travel + /// (`AT`/`BEFORE`) and options are not lineage-relevant and are skipped. + /// See: + CreateStream { + or_replace: bool, + if_not_exists: bool, + name: WithSpan, + source_type: String, + source: WithSpan, + }, + /// ```sql + /// CREATE [ OR REPLACE ] PIPE [ IF NOT EXISTS ] + /// [ ... ] AS + /// ``` + /// + /// Snowflake pipe (Snowpipe). The `AS ` body is a + /// `COPY INTO FROM ` statement, so it carries the ingest + /// lineage edge `stage -> table`. `body` is `None` only when the body + /// could not be parsed. + /// See: + CreatePipe { + or_replace: bool, + if_not_exists: bool, + name: WithSpan, + body: Option>, + }, + /// ```sql + /// CREATE [ OR REPLACE ] MODEL [ IF NOT EXISTS ] + /// [ TRANSFORM ( ... ) ] [ OPTIONS ( ... ) ] AS -- BigQuery ML + /// CREATE MODEL FROM {
| ( ) } ... -- Redshift ML + /// ``` + /// + /// BigQuery ML / Redshift ML model. The training input is the lineage edge + /// `training_source -> model`, captured as either `query` (BigQuery + /// `AS ` or Redshift `FROM ()`) or `from_table` (Redshift + /// `FROM
`). At most one of the two is set. `TRANSFORM`, `OPTIONS`, + /// `TARGET`, `FUNCTION`, `IAM_ROLE` and `SETTINGS` are not lineage-relevant + /// and are skipped. + /// See: + /// and + CreateModel { + or_replace: bool, + if_not_exists: bool, + name: WithSpan, + /// The training-input query: BigQuery `AS ` or Redshift ML + /// `FROM ()`. Mutually exclusive with `from_table`. + query: Option>, + /// The training-input table for a Redshift ML `FROM
` source + /// (when it isn't a subquery). Mutually exclusive with `query`. + from_table: Option>, + }, + /// ```sql + /// CREATE EXTERNAL SCHEMA [ IF NOT EXISTS ] + /// FROM { DATA CATALOG | HIVE METASTORE | REDSHIFT | POSTGRES | MYSQL | ... } + /// [ DATABASE '' [ SCHEMA '' ] ] [ ... ] + /// ``` + /// + /// Redshift external schema. Maps a local schema name onto an external + /// metastore / federated database, so it carries the lineage edge from the + /// local schema to the external `from` source identified by `database` / + /// `schema`. Region / URI / IAM role options are not lineage-relevant and + /// are skipped. + /// See: + CreateExternalSchema { + if_not_exists: bool, + name: WithSpan, + /// The `FROM` source kind, uppercased (e.g. `DATA CATALOG`, + /// `HIVE METASTORE`, `REDSHIFT`, `POSTGRES`, `MYSQL`, `KINESIS`). + from: String, + /// `DATABASE ''` — the external database this schema maps to. + database: Option, + /// `SCHEMA ''` — the external schema (Redshift federated source). + schema: Option, + }, + /// ```sql + /// CREATE ... + /// ``` + /// + /// A `CREATE` statement whose object type this parser does not model (e.g. + /// `CREATE WAREHOUSE`, `CREATE FILE FORMAT`, `CREATE DICTIONARY`). The + /// remainder of the statement is captured verbatim in `body` rather than + /// being represented structurally. This exists so consumers can distinguish + /// a genuinely unsupported `CREATE` from a real statement — previously these + /// were misrepresented as a no-op `COMMENT ON `. + CreateUnsupported { + or_replace: bool, + /// The object-type keyword that followed `CREATE`, uppercased + /// (e.g. `WAREHOUSE`, `FILE`). + object_type: String, + /// The raw text consumed after `object_type` up to the statement end, + /// preserved verbatim for inspection / re-parsing. Empty if nothing + /// followed the object type. + body: String, + }, + /// ```sql + /// DROP ... + /// ``` + /// + /// A `DROP` statement whose object type this parser does not model (e.g. + /// `DROP WAREHOUSE`, `DROP XML SCHEMA COLLECTION`). The remainder is captured + /// verbatim in `body`. Lets consumers skip an unmodelled drop without failing + /// the surrounding multi-statement block. + DropUnsupported { + /// The object-type keyword that followed `DROP`, uppercased. + object_type: String, + /// The raw text consumed after `object_type` up to the statement end. + body: String, + }, + /// ```sql + /// ALTER ... + /// ``` + /// + /// An `ALTER` statement whose object type this parser does not model (e.g. + /// `ALTER WAREHOUSE`, `ALTER TASK`). The remainder is captured verbatim in + /// `body`. + AlterUnsupported { + /// The object-type keyword that followed `ALTER`, uppercased. + object_type: String, + /// The raw text consumed after `object_type` up to the statement end. + body: String, + }, + /// A statement whose leading keyword is recognised but whose body this parser + /// does not model structurally (e.g. ClickHouse `SYSTEM ...`, PostgreSQL + /// `LOCK TABLE ...`). The body is captured verbatim in `body`. Previously + /// these were misrepresented as a no-op `SET `. + UnsupportedStatement { + /// The leading keyword, uppercased (e.g. `SYSTEM`, `LOCK`). + keyword: String, + /// The raw text consumed after `keyword` up to the statement end. + body: String, + }, + /// ```sql + /// DO [ LANGUAGE ] + /// ``` + /// + /// PostgreSQL anonymous code block — runs a procedural body once without + /// creating a named routine. The `body` is the code string (typically + /// dollar-quoted `$$ ... $$`), captured as a re-parseable + /// [`FunctionDefinition`] with its source start location preserved, so + /// downstream consumers (lineage extraction) can re-parse it for embedded + /// DML/DDL. The optional `LANGUAGE` clause is consumed but not represented. + /// See: + Do { body: Option }, + /// ```sql /// CREATE MACRO /// ``` /// @@ -2757,7 +2937,7 @@ pub enum Statement { /// /// Snowflake-specific statement that triggers an ad-hoc run of a task. /// See: - ExecuteTask { name: ObjectName }, + ExecuteTask { name: WithSpan }, /// ```sql /// EXPORT DATA [WITH CONNECTION conn] OPTIONS ( ) AS /// ``` @@ -2780,7 +2960,16 @@ pub enum Statement { /// matters for lineage; everything after `FROM FILES` is captured as /// opaque text since FROM FILES options carry no table refs. /// See: + /// + /// Also covers the Hive/Spark/Databricks form + /// `LOAD DATA [LOCAL] INPATH '' [OVERWRITE] INTO TABLE [PARTITION ...]`, + /// where `inpath` is the file source and `target` is the loaded table. + /// See: LoadData { + /// Hive `LOCAL` modifier (load from the client filesystem). + local: bool, + /// Hive `INPATH ''` file source, if present. + inpath: Option, overwrite: bool, target: ObjectName, rest: String, @@ -3293,7 +3482,7 @@ impl fmt::Display for Statement { extension_name: name, } => write!(f, "INSTALL {name}"), - Statement::Load { + Statement::DuckDbLoad { extension_name: name, } => write!(f, "LOAD {name}"), @@ -3455,6 +3644,160 @@ impl fmt::Display for Statement { write!(f, "{params}")?; Ok(()) } + Statement::CreateTask { + or_replace, + or_alter, + if_not_exists, + name, + after, + when, + body, + } => { + write!(f, "CREATE ")?; + if *or_replace { + write!(f, "OR REPLACE ")?; + } else if *or_alter { + write!(f, "OR ALTER ")?; + } + write!(f, "TASK ")?; + if *if_not_exists { + write!(f, "IF NOT EXISTS ")?; + } + write!(f, "{name}")?; + if !after.is_empty() { + write!(f, " AFTER {}", display_comma_separated(after))?; + } + if let Some(when) = when { + write!(f, " WHEN {when}")?; + } + if let Some(body) = body { + write!(f, " AS {body}")?; + } + Ok(()) + } + Statement::CreateStream { + or_replace, + if_not_exists, + name, + source_type, + source, + } => { + write!(f, "CREATE ")?; + if *or_replace { + write!(f, "OR REPLACE ")?; + } + write!(f, "STREAM ")?; + if *if_not_exists { + write!(f, "IF NOT EXISTS ")?; + } + write!(f, "{name} ON {source_type} {source}") + } + Statement::CreatePipe { + or_replace, + if_not_exists, + name, + body, + } => { + write!(f, "CREATE ")?; + if *or_replace { + write!(f, "OR REPLACE ")?; + } + write!(f, "PIPE ")?; + if *if_not_exists { + write!(f, "IF NOT EXISTS ")?; + } + write!(f, "{name}")?; + if let Some(body) = body { + write!(f, " AS {body}")?; + } + Ok(()) + } + Statement::CreateModel { + or_replace, + if_not_exists, + name, + query, + from_table, + } => { + write!(f, "CREATE ")?; + if *or_replace { + write!(f, "OR REPLACE ")?; + } + write!(f, "MODEL ")?; + if *if_not_exists { + write!(f, "IF NOT EXISTS ")?; + } + write!(f, "{name}")?; + if let Some(query) = query { + write!(f, " AS {query}")?; + } else if let Some(from_table) = from_table { + write!(f, " FROM {from_table}")?; + } + Ok(()) + } + Statement::CreateExternalSchema { + if_not_exists, + name, + from, + database, + schema, + } => { + write!(f, "CREATE EXTERNAL SCHEMA ")?; + if *if_not_exists { + write!(f, "IF NOT EXISTS ")?; + } + write!(f, "{name} FROM {from}")?; + if let Some(database) = database { + write!(f, " DATABASE '{database}'")?; + } + if let Some(schema) = schema { + write!(f, " SCHEMA '{schema}'")?; + } + Ok(()) + } + Statement::CreateUnsupported { + or_replace, + object_type, + body, + } => { + write!(f, "CREATE ")?; + if *or_replace { + write!(f, "OR REPLACE ")?; + } + write!(f, "{object_type}")?; + if !body.is_empty() { + write!(f, " {body}")?; + } + Ok(()) + } + Statement::DropUnsupported { object_type, body } => { + write!(f, "DROP {object_type}")?; + if !body.is_empty() { + write!(f, " {body}")?; + } + Ok(()) + } + Statement::AlterUnsupported { object_type, body } => { + write!(f, "ALTER {object_type}")?; + if !body.is_empty() { + write!(f, " {body}")?; + } + Ok(()) + } + Statement::UnsupportedStatement { keyword, body } => { + write!(f, "{keyword}")?; + if !body.is_empty() { + write!(f, " {body}")?; + } + Ok(()) + } + Statement::Do { body } => { + write!(f, "DO")?; + if let Some(body) = body { + write!(f, " {body}")?; + } + Ok(()) + } Statement::CreateProcedure { name, or_alter, @@ -4888,17 +5231,33 @@ impl fmt::Display for Statement { write!(f, "EXPORT DATA OPTIONS ({options}) AS {query}") } Statement::LoadData { + local, + inpath, overwrite, target, rest, } => { write!(f, "LOAD DATA ")?; - if *overwrite { - write!(f, "OVERWRITE ")?; + if let Some(inpath) = inpath { + // Hive/Spark form: LOAD DATA [LOCAL] INPATH '' + // [OVERWRITE] INTO TABLE . + if *local { + write!(f, "LOCAL ")?; + } + write!(f, "INPATH '{inpath}' ")?; + if *overwrite { + write!(f, "OVERWRITE ")?; + } + write!(f, "INTO TABLE {target}")?; } else { - write!(f, "INTO ")?; + // BigQuery form: LOAD DATA { OVERWRITE | INTO } . + if *overwrite { + write!(f, "OVERWRITE ")?; + } else { + write!(f, "INTO ")?; + } + write!(f, "{target}")?; } - write!(f, "{target}")?; if !rest.is_empty() { write!(f, " {rest}")?; } diff --git a/src/parser/mod.rs b/src/parser/mod.rs index 455eb0ae5..0b6d2f83b 100644 --- a/src/parser/mod.rs +++ b/src/parser/mod.rs @@ -930,28 +930,42 @@ impl<'a> Parser<'a> { Keyword::INSTALL if dialect_of!(self is DuckDbDialect | GenericDialect) => { Ok(self.parse_install()?) } - // BigQuery `LOAD DATA [OVERWRITE | INTO] ... FROM FILES (...)`. - // Recognize before the DuckDB `LOAD ` branch so the + // `LOAD DATA ...` in two shapes, both ending at a target table: + // BigQuery: LOAD DATA [OVERWRITE | INTO] ... FROM FILES (...) + // Hive/Spark/Databricks: + // LOAD DATA [LOCAL] INPATH '' [OVERWRITE] INTO TABLE [PARTITION ...] + // Recognized before the DuckDB `LOAD ` branch so the // `DATA` keyword distinguishes the two forms. Keyword::LOAD - if dialect_of!(self is BigQueryDialect | GenericDialect) + if dialect_of!(self is BigQueryDialect | GenericDialect | HiveDialect | DatabricksDialect) && matches!( self.peek_token_kind(), Token::Word(w) if w.keyword == Keyword::DATA ) => { self.expect_keyword(Keyword::DATA)?; - let overwrite = if self.parse_keyword(Keyword::OVERWRITE) { - true + // Hive `LOCAL INPATH ''` source (INPATH is not a + // reserved keyword, so match it by value). + let local = self.parse_keyword(Keyword::LOCAL); + let inpath = if matches!( + self.peek_token_kind(), + Token::Word(w) if w.value.eq_ignore_ascii_case("INPATH") + ) { + self.next_token(); // INPATH + Some(self.parse_literal_string()?) } else { - self.expect_keyword(Keyword::INTO)?; - false + None }; + let overwrite = self.parse_keyword(Keyword::OVERWRITE); + // `INTO` is present in both Hive (`INTO TABLE`) and BigQuery + // (`INTO `); BigQuery's `OVERWRITE` form omits it. + let _ = self.parse_keyword(Keyword::INTO); + let _ = self.parse_keyword(Keyword::TABLE); // Hive: INTO TABLE let target = self.parse_object_name(true)?; // Everything from here to end-of-statement is opaque // (column list, OPTIONS, FROM FILES, WITH PARTITION - // COLUMNS …). No further lineage payload — `target` is - // the only output reference. + // COLUMNS, PARTITION (...) …). No further lineage payload — + // `target` is the output and `inpath` the source. let start = self.index; while !self.peek_token_is(&Token::EOF) && !self.peek_token_is(&Token::SemiColon) { @@ -959,10 +973,13 @@ impl<'a> Parser<'a> { } let rest: String = self.tokens[start..self.index] .iter() + .filter(|t| !matches!(t.token, Token::Whitespace(_))) .map(|t| t.token.to_string()) .collect::>() .join(" "); Ok(Statement::LoadData { + local, + inpath, overwrite, target, rest, @@ -977,52 +994,30 @@ impl<'a> Parser<'a> { Ok(self.parse_optimize_table()?) } // ClickHouse SYSTEM statements (SYSTEM RELOAD, SYSTEM RESTORE, etc.) + // Not modelled structurally; capture the body verbatim rather + // than misrepresenting it as a `SET SYSTEM` statement. Keyword::SYSTEM if dialect_of!(self is ClickHouseDialect | GenericDialect) => { - // Skip remaining tokens until end of statement - while !self.peek_token_is(&Token::EOF) && !self.peek_token_is(&Token::SemiColon) - { - self.next_token(); - } - Ok(Statement::SetVariable { - local: false, - hivevar: false, - tuple: false, - variable: ObjectName(vec!["SYSTEM".into()]), - value: vec![], - additional_assignments: vec![], + let body = self.consume_statement_body_text(); + Ok(Statement::UnsupportedStatement { + keyword: "SYSTEM".to_string(), + body, }) } // PostgreSQL LOCK TABLE: LOCK TABLE IN MODE; + // Captured verbatim (the table reference is preserved in the body + // text) instead of being misrepresented as a `SET LOCK` statement. Keyword::LOCK if dialect_of!(self is PostgreSqlDialect | GenericDialect) => { - // Skip remaining tokens until end of statement - while !self.peek_token_is(&Token::EOF) && !self.peek_token_is(&Token::SemiColon) - { - self.next_token(); - } - Ok(Statement::SetVariable { - local: false, - hivevar: false, - tuple: false, - variable: ObjectName(vec!["LOCK".into()]), - value: vec![], - additional_assignments: vec![], + let body = self.consume_statement_body_text(); + Ok(Statement::UnsupportedStatement { + keyword: "LOCK".to_string(), + body, }) } - // PostgreSQL DO block: DO $$ ... $$; + // PostgreSQL anonymous code block: DO [LANGUAGE ] $$ ... $$; + // The procedural body is captured as a re-parseable + // FunctionDefinition so embedded DML/DDL stays recoverable. Keyword::DO if dialect_of!(self is PostgreSqlDialect | GenericDialect) => { - // Skip remaining tokens until end of statement - while !self.peek_token_is(&Token::EOF) && !self.peek_token_is(&Token::SemiColon) - { - self.next_token(); - } - Ok(Statement::SetVariable { - local: false, - hivevar: false, - tuple: false, - variable: ObjectName(vec!["DO".into()]), - value: vec![], - additional_assignments: vec![], - }) + self.parse_do() } // CASE as a standalone expression statement (e.g. Snowflake masking policy bodies) Keyword::CASE if dialect_of!(self is SnowflakeDialect | BigQueryDialect) => { @@ -4788,6 +4783,9 @@ impl<'a> Parser<'a> { pub fn parse_create(&mut self) -> Result { let or_replace = self.parse_keywords(&[Keyword::OR, Keyword::REPLACE]); let or_alter = self.parse_keywords(&[Keyword::OR, Keyword::ALTER]); + // Databricks: `CREATE OR REFRESH STREAMING TABLE | MATERIALIZED VIEW ...`. + // REFRESH is not a replace/alter; consume it as a no-op modifier. + let _or_refresh = self.parse_keywords(&[Keyword::OR, Keyword::REFRESH]); let local = self.parse_one_of_keywords(&[Keyword::LOCAL]).is_some(); let global = self.parse_one_of_keywords(&[Keyword::GLOBAL]).is_some(); // Snowflake `SECURE` modifier: precedes VIEW / MATERIALIZED VIEW / FUNCTION. @@ -4869,6 +4867,10 @@ impl<'a> Parser<'a> { if self.parse_keyword(Keyword::FUNCTION) { // Snowflake: CREATE EXTERNAL FUNCTION self.parse_create_function_inner(or_replace, temporary, false, secure) + } else if self.parse_keyword(Keyword::SCHEMA) { + // Redshift: CREATE EXTERNAL SCHEMA FROM { DATA CATALOG | + // HIVE METASTORE | REDSHIFT | POSTGRES | ... } DATABASE '...' + self.parse_create_external_schema() } else { self.parse_create_external_table(or_replace) } @@ -4941,42 +4943,81 @@ impl<'a> Parser<'a> { // BigQuery row-level security: // ROW ACCESS POLICY ON
[GRANT TO (...)] FILTER USING (expr) self.parse_create_row_access_policy(or_replace) + } else if self.parse_keyword(Keyword::TASK) { + // Snowflake: CREATE [OR REPLACE | OR ALTER] TASK ... AS + self.parse_create_task(or_replace, or_alter) + } else if self.parse_keyword(Keyword::STREAM) { + // Snowflake: CREATE [OR REPLACE] STREAM ... ON TABLE/VIEW/STAGE + self.parse_create_stream(or_replace) + } else if self.peek_word_ci("PIPE") { + // Snowflake: CREATE [OR REPLACE] PIPE ... AS . + // PIPE is not a reserved keyword, so match it by value. + self.next_token(); // PIPE + self.parse_create_pipe(or_replace) + } else if self.parse_keyword(Keyword::MODEL) { + // BigQuery ML: CREATE [OR REPLACE] MODEL ... AS + self.parse_create_model(or_replace) + } else if self.parse_keywords(&[Keyword::FOREIGN, Keyword::TABLE]) { + // PostgreSQL foreign table — an external-table source. Capture the + // name and column list (the lineage-relevant schema) and skip the + // trailing `SERVER OPTIONS (...)` clause, which the shared + // external-table parser's option grammar doesn't recognise. + let if_not_exists = self.parse_keywords(&[Keyword::IF, Keyword::NOT, Keyword::EXISTS]); + let table_name = self.parse_object_name(false)?; + let (columns, constraints, _) = self.parse_columns()?; + self.skip_to_statement_end(); + Ok(CreateTableBuilder::new(table_name) + .columns(columns) + .constraints(constraints) + .or_replace(or_replace) + .if_not_exists(if_not_exists) + .external(true) + .build()) + } else if self.peek_word_ci("STREAMING") + && matches!(self.peek_nth_token(1).token, Token::Word(w) if w.keyword == Keyword::TABLE) + { + // Databricks: CREATE [OR REFRESH] STREAMING TABLE [...] AS . + // Route a plain `AS ` form through the regular table parser so + // the query is captured for lineage. Databricks Delta Live Tables add + // FLOW / WATERMARK / EXPECT-constraint / SCD clauses we don't model; + // if the table parse can't consume the whole statement, fall back to + // CreateUnsupported rather than erroring. + self.next_token(); // STREAMING + self.expect_keyword(Keyword::TABLE)?; + let start = self.index; + match self.maybe_parse(|p| { + p.parse_create_table_inner( + or_replace, temporary, global, transient, dynamic, iceberg, hybrid, + ) + }) { + Some(stmt) if matches!(self.peek_token_kind(), Token::EOF | Token::SemiColon) => { + Ok(stmt) + } + _ => { + self.index = start; + let body = self.consume_statement_body_text(); + Ok(Statement::CreateUnsupported { + or_replace, + object_type: "STREAMING TABLE".to_string(), + body, + }) + } + } } else { - // Generic fallback: skip tokens until end of statement - // This handles dialect-specific CREATE statements like: - // CREATE DICTIONARY, CREATE WAREHOUSE, CREATE DYNAMIC TABLE, - // CREATE USER, CREATE STORAGE, CREATE FILE FORMAT, etc. + // Generic fallback: this CREATE is not modelled. + // Capture the object type and the remaining text verbatim and return + // a CreateUnsupported node so consumers can tell it apart from a real + // statement. Examples: CREATE WAREHOUSE, CREATE FILE FORMAT, + // CREATE DICTIONARY, CREATE USER, CREATE STORAGE INTEGRATION, etc. let token = self.next_token(); match token.token { Token::Word(w) => { let object_type = w.value.to_uppercase(); - // Skip remaining tokens until semicolon or EOF - let mut depth = 0i32; - loop { - match self.peek_token_kind().clone() { - Token::EOF => break, - Token::SemiColon if depth == 0 => break, - Token::LParen => { - depth += 1; - self.next_token(); - } - Token::RParen => { - if depth == 0 { - break; - } - depth -= 1; - self.next_token(); - } - _ => { - self.next_token(); - } - } - } - Ok(Statement::Comment { - object_type: CommentObject::Other(object_type), - object_name: ObjectName(vec![]), - comment: None, - if_exists: false, + let body = self.consume_statement_body_text(); + Ok(Statement::CreateUnsupported { + or_replace, + object_type, + body, }) } _ => self.expected("an object type after CREATE", token), @@ -4984,6 +5025,388 @@ impl<'a> Parser<'a> { } } + /// Parse a Snowflake `CREATE [OR REPLACE | OR ALTER] TASK [IF NOT EXISTS] + /// [] [AFTER [, ...]] [WHEN ] AS `. + /// + /// The `TASK` keyword has already been consumed by the caller. Scheduling, + /// warehouse and session properties are not lineage-relevant and are + /// consumed token-by-token; only the task name, `AFTER` predecessors, the + /// `WHEN` predicate and the `AS ` body are retained. + /// + fn parse_create_task( + &mut self, + or_replace: bool, + or_alter: bool, + ) -> Result { + let if_not_exists = self.parse_keywords(&[Keyword::IF, Keyword::NOT, Keyword::EXISTS]); + let name_start = self.index; + let name = self.parse_object_name(false)?; + let name = name.spanning(self.span_from_index(name_start)); + + let mut after = vec![]; + let mut when = None; + let mut found_as = false; + loop { + match self.peek_token_kind() { + Token::EOF | Token::SemiColon => break, + _ => {} + } + if self.parse_keyword(Keyword::AS) { + found_as = true; + break; + } + // `EXECUTE AS { CALLER | OWNER | SELF | USER }` — consume the + // `EXECUTE AS` pair so its `AS` is not mistaken for the body marker. + if self.parse_keywords(&[Keyword::EXECUTE, Keyword::AS]) { + continue; + } + if self.parse_keyword(Keyword::AFTER) { + // AFTER [, ...] + loop { + let dep_start = self.index; + let dep = self.parse_object_name(false)?; + after.push(dep.spanning(self.span_from_index(dep_start))); + if !self.consume_token(&Token::Comma) { + break; + } + } + continue; + } + if self.parse_keyword(Keyword::WHEN) { + let cond_start = self.index; + let cond = self.parse_expr()?; + when = Some(cond.spanning(self.span_from_index(cond_start))); + continue; + } + // Opaque property token (WAREHOUSE = ..., SCHEDULE = ..., FINALIZE = + // ..., session params, etc.). + self.next_token(); + } + + // Parse the body as a single statement (INSERT / MERGE / CALL / + // BEGIN..END block / EXECUTE IMMEDIATE / ...), falling back gracefully + // for shapes that can't be represented (e.g. `EXECUTE DBT PROJECT`). + let body = if found_as { + self.parse_create_body_statement() + } else { + None + }; + + Ok(Statement::CreateTask { + or_replace, + or_alter, + if_not_exists, + name, + after, + when, + body, + }) + } + + /// Parse a Snowflake `CREATE [OR REPLACE] STREAM [IF NOT EXISTS] + /// [COPY GRANTS] ON [{AT|BEFORE}(...)] []`. + /// + /// The `STREAM` keyword has already been consumed by the caller. The lineage + /// edge is `source -> stream`; time-travel and options are skipped. + /// + fn parse_create_stream(&mut self, or_replace: bool) -> Result { + let if_not_exists = self.parse_keywords(&[Keyword::IF, Keyword::NOT, Keyword::EXISTS]); + let name_start = self.index; + let name = self.parse_object_name(false)?; + let name = name.spanning(self.span_from_index(name_start)); + + // Skip pre-ON properties (COPY GRANTS, etc.) until the `ON` source clause. + let mut found_on = false; + loop { + match self.peek_token_kind() { + Token::EOF | Token::SemiColon => break, + Token::Word(w) if w.keyword == Keyword::ON => { + self.next_token(); + found_on = true; + break; + } + _ => { + self.next_token(); + } + } + } + + let (source_type, source) = if found_on { + // Collect the object-kind keyword(s): optional modifiers + // (EXTERNAL/DIRECTORY/DYNAMIC/ICEBERG) then a noun (TABLE/VIEW/STAGE). + let mut source_type = String::new(); + while let Token::Word(w) = self.peek_token_kind() { + let word = w.value.to_uppercase(); + match word.as_str() { + "EXTERNAL" | "DIRECTORY" | "DYNAMIC" | "ICEBERG" => { + if !source_type.is_empty() { + source_type.push(' '); + } + source_type.push_str(&word); + self.next_token(); + } + "TABLE" | "VIEW" | "STAGE" => { + if !source_type.is_empty() { + source_type.push(' '); + } + source_type.push_str(&word); + self.next_token(); + break; + } + _ => break, + } + } + let src_start = self.index; + let source = self.parse_object_name(false)?; + let source = source.spanning(self.span_from_index(src_start)); + (source_type, source) + } else { + (String::new(), ObjectName(vec![]).empty_span()) + }; + + // Skip the remainder (AT/BEFORE time-travel, APPEND_ONLY, COMMENT, ...). + self.skip_to_statement_end(); + + Ok(Statement::CreateStream { + or_replace, + if_not_exists, + name, + source_type, + source, + }) + } + + /// Parse a Snowflake `CREATE [OR REPLACE] PIPE [IF NOT EXISTS] + /// [] AS `. + /// + /// The `PIPE` keyword has already been consumed by the caller. The `AS` + /// body is a `COPY INTO` statement carrying ingest lineage (stage -> table). + /// + fn parse_create_pipe(&mut self, or_replace: bool) -> Result { + let if_not_exists = self.parse_keywords(&[Keyword::IF, Keyword::NOT, Keyword::EXISTS]); + let name_start = self.index; + let name = self.parse_object_name(false)?; + let name = name.spanning(self.span_from_index(name_start)); + + // Skip pre-AS properties (AUTO_INGEST, AWS_SNS_TOPIC, INTEGRATION, + // ERROR_INTEGRATION, COMMENT, ...) until the AS body marker. + let mut found_as = false; + loop { + match self.peek_token_kind() { + Token::EOF | Token::SemiColon => break, + _ => {} + } + if self.parse_keyword(Keyword::AS) { + found_as = true; + break; + } + self.next_token(); + } + + let body = if found_as { + self.parse_create_body_statement() + } else { + None + }; + + Ok(Statement::CreatePipe { + or_replace, + if_not_exists, + name, + body, + }) + } + + /// Parse a BigQuery ML `CREATE [OR REPLACE] MODEL [IF NOT EXISTS] + /// [TRANSFORM(...)] [OPTIONS(...)] AS `. + /// + /// The `MODEL` keyword has already been consumed by the caller. The `AS` + /// query is the training input; `TRANSFORM` and `OPTIONS` are skipped. + /// + fn parse_create_model(&mut self, or_replace: bool) -> Result { + let if_not_exists = self.parse_keywords(&[Keyword::IF, Keyword::NOT, Keyword::EXISTS]); + let name_start = self.index; + let name = self.parse_object_name(false)?; + let name = name.spanning(self.span_from_index(name_start)); + + // Find the training-input body. BigQuery uses `AS ` (preceded by + // optional TRANSFORM(...)/OPTIONS(...)); Redshift ML uses `FROM ()` + // or `FROM
` (followed by TARGET/FUNCTION/IAM_ROLE/SETTINGS). + // Skip balanced-paren option clauses until we hit AS or FROM. + let mut marker = None; + loop { + match self.peek_token_kind() { + Token::EOF | Token::SemiColon => break, + _ => {} + } + if self.parse_keyword(Keyword::AS) { + marker = Some(Keyword::AS); + break; + } + if self.parse_keyword(Keyword::FROM) { + marker = Some(Keyword::FROM); + break; + } + if self.consume_token(&Token::LParen) { + let mut depth = 1i32; + while depth > 0 { + match self.next_token().token { + Token::LParen => depth += 1, + Token::RParen => depth -= 1, + Token::EOF => break, + _ => {} + } + } + continue; + } + self.next_token(); + } + + let mut query = None; + let mut from_table = None; + match marker { + // BigQuery: AS . + Some(Keyword::AS) => query = Some(Box::new(self.parse_query()?)), + Some(Keyword::FROM) => { + if self.consume_token(&Token::LParen) { + // Redshift ML: FROM () training input. + let q = self.parse_query()?; + self.expect_token(&Token::RParen)?; + query = Some(Box::new(q)); + } else { + // Redshift ML: FROM
training input. + let start = self.index; + let table = self.parse_object_name(false)?; + from_table = Some(table.spanning(self.span_from_index(start))); + } + } + _ => {} + } + // Discard trailing clauses (TARGET/FUNCTION/IAM_ROLE/SETTINGS/...). + self.skip_to_statement_end(); + Ok(Statement::CreateModel { + or_replace, + if_not_exists, + name, + query, + from_table, + }) + } + + /// Parse a Redshift `CREATE EXTERNAL SCHEMA [IF NOT EXISTS] FROM + /// { DATA CATALOG | HIVE METASTORE | REDSHIFT | POSTGRES | MYSQL | ... } + /// [DATABASE '' [SCHEMA '']] []`. + /// + /// `EXTERNAL SCHEMA` has already been consumed by the caller. The local + /// schema name plus the federated `FROM` source (database/schema) are + /// retained for lineage; region / URI / IAM-role options are skipped. + /// + fn parse_create_external_schema(&mut self) -> Result { + let if_not_exists = self.parse_keywords(&[Keyword::IF, Keyword::NOT, Keyword::EXISTS]); + let name_start = self.index; + let name = self.parse_object_name(false)?; + let name = name.spanning(self.span_from_index(name_start)); + self.expect_keyword(Keyword::FROM)?; + + // FROM source kind: a single word, or the two-word forms + // `DATA CATALOG` / `HIVE METASTORE`. + let first = self.next_token(); + let mut from = match first.token { + Token::Word(w) => w.value.to_uppercase(), + _ => return self.expected("an external schema source after FROM", first), + }; + if from == "DATA" && self.peek_word_ci("CATALOG") { + self.next_token(); + from = "DATA CATALOG".to_string(); + } else if from == "HIVE" && self.peek_word_ci("METASTORE") { + self.next_token(); + from = "HIVE METASTORE".to_string(); + } + + // The external database / schema this maps to (lineage target). + let database = if self.parse_keyword(Keyword::DATABASE) { + Some(self.parse_literal_string()?) + } else { + None + }; + let schema = if self.parse_keyword(Keyword::SCHEMA) { + Some(self.parse_literal_string()?) + } else { + None + }; + + // Skip the remainder (REGION/URI/PORT/IAM_ROLE/...). + self.skip_to_statement_end(); + + Ok(Statement::CreateExternalSchema { + if_not_exists, + name, + from, + database, + schema, + }) + } + + /// Parse the `AS ` body of a CREATE TASK / CREATE PIPE as a single + /// statement. Returns `None` (skipping to the statement end) when the body + /// can't be parsed, or when it only *partially* parses — e.g. Snowflake's + /// `EXECUTE DBT PROJECT ...` parses as `EXECUTE DBT` and leaves trailing + /// tokens, which would otherwise corrupt the surrounding statement. + fn parse_create_body_statement(&mut self) -> Option> { + let start = self.index; + if let Some(stmt) = self.maybe_parse(|p| p.parse_statement()) { + if matches!(self.peek_token_kind(), Token::EOF | Token::SemiColon) { + return Some(Box::new(stmt)); + } + } + // Unparseable or partial body: revert and discard the remainder. + self.index = start; + self.skip_to_statement_end(); + None + } + + /// Consume tokens up to (but not including) the statement-terminating + /// semicolon or EOF, respecting balanced parentheses. Used by the + /// lineage-focused CREATE parsers to discard trailing option clauses. + fn skip_to_statement_end(&mut self) { + let mut depth = 0i32; + loop { + match self.peek_token_kind() { + Token::EOF => break, + Token::SemiColon if depth == 0 => break, + Token::LParen => { + depth += 1; + self.next_token(); + } + Token::RParen => { + if depth == 0 { + break; + } + depth -= 1; + self.next_token(); + } + _ => { + self.next_token(); + } + } + } + } + + /// Consume tokens to the end of the statement (respecting parens) and return + /// the raw text consumed, joined with single spaces. Used by the opaque + /// `*Unsupported` statement nodes to preserve the unmodelled body verbatim + /// for inspection / re-parsing. + fn consume_statement_body_text(&mut self) -> String { + let start = self.index; + self.skip_to_statement_end(); + self.tokens[start..self.index] + .iter() + .filter(|t| !matches!(t.token, Token::Whitespace(_))) + .map(|t| t.token.to_string()) + .collect::>() + .join(" ") + } + /// Parse a Snowflake security/governance policy *definition*: /// `{ MASKING | ROW ACCESS | AGGREGATION | PROJECTION | JOIN } POLICY /// [IF NOT EXISTS] AS ( [ , ...] ) RETURNS -> @@ -6991,6 +7414,37 @@ impl<'a> Parser<'a> { } } + // Materialized views in Oracle / Databricks / Trino / BigQuery carry + // assorted pre-AS option clauses that aren't modelled above and don't + // carry lineage: Oracle `BUILD ... REFRESH ... ON {COMMIT|DEMAND}`, + // `PCTFREE n`, Databricks `TRIGGER ON UPDATE` / `SCHEDULE {EVERY ...| + // CRON '...' AT TIME ZONE '...'}`, Trino `GRACE PERIOD ...`, BigQuery + // `ANNOTATIONS ...`, etc. Skip any leftover tokens (respecting parens) + // up to the `AS` query body so the lineage-bearing query is preserved. + // ClickHouse is excluded: its `[APPEND] TO
` output target and + // ENGINE/REFRESH clauses are parsed precisely above, and a blind skip + // would drop the `TO
` output edge. + if materialized && !dialect_of!(self is ClickHouseDialect) { + let mut depth = 0i32; + loop { + match self.peek_token_kind() { + Token::EOF | Token::SemiColon => break, + Token::Word(w) if depth == 0 && w.keyword == Keyword::AS => break, + Token::LParen => { + depth += 1; + self.next_token(); + } + Token::RParen => { + depth = depth.saturating_sub(1); + self.next_token(); + } + _ => { + self.next_token(); + } + } + } + } + self.expect_keyword(Keyword::AS)?; let query = self.parse_boxed_query()?; // Optional `WITH [ CASCADED | LOCAL ] CHECK OPTION` (Postgres/MySQL view updatability constraint). @@ -7376,6 +7830,14 @@ impl<'a> Parser<'a> { return self.parse_drop_function(); } else if self.parse_keyword(Keyword::PROCEDURE) { return self.parse_drop_function(); + } else if let Token::Word(w) = self.peek_token_kind().clone() { + // Unmodelled DROP object type (e.g. DROP WAREHOUSE, DROP STREAM, + // DROP PIPE, DROP TASK). Capture it opaquely so a multi-statement + // block isn't failed by a single unsupported drop. + self.next_token(); + let object_type = w.value.to_uppercase(); + let body = self.consume_statement_body_text(); + return Ok(Statement::DropUnsupported { object_type, body }); } else { return self.expected( "TABLE, VIEW, INDEX, ROLE, SCHEMA, DATABASE, FUNCTION, PROCEDURE, STAGE or SEQUENCE after DROP", @@ -10702,14 +11164,31 @@ impl<'a> Parser<'a> { if self.parse_keyword(Keyword::POLICY) { return self.parse_alter_postgres_policy(); } - let object_type = self.expect_one_of_keywords(&[ + let object_type = self.parse_one_of_keywords(&[ Keyword::VIEW, Keyword::TABLE, Keyword::INDEX, Keyword::ROLE, Keyword::SCHEMA, Keyword::SESSION, - ])?; + ]); + let object_type = match object_type { + Some(kw) => kw, + // Unmodelled ALTER target (e.g. ALTER WAREHOUSE, ALTER TASK, + // ALTER STREAM). Capture it opaquely rather than failing. + None => { + if let Token::Word(w) = self.peek_token_kind().clone() { + self.next_token(); + let object_type = w.value.to_uppercase(); + let body = self.consume_statement_body_text(); + return Ok(Statement::AlterUnsupported { object_type, body }); + } + return self.expected( + "VIEW, TABLE, INDEX, ROLE, SCHEMA or SESSION after ALTER", + self.peek_token(), + ); + } + }; match object_type { Keyword::VIEW => self.parse_alter_view(), Keyword::TABLE => { @@ -11821,7 +12300,9 @@ impl<'a> Parser<'a> { )))) } } - Keyword::MAP if dialect_of!(self is DatabricksDialect) => { + Keyword::MAP if dialect_of!(self is DatabricksDialect | RedshiftSqlDialect) => { + // Databricks `MAP` and Redshift Spectrum (Hive-style) + // `map` both use angle-bracket, comma-separated types. self.prev_token(); let (field_defs, _trailing_bracket) = self.parse_struct_type_def(Self::parse_struct_field_def, Keyword::MAP)?; @@ -11835,7 +12316,10 @@ impl<'a> Parser<'a> { trailing_bracket = _trailing_bracket; Ok(DataType::Struct(field_defs)) } - Keyword::STRUCT if dialect_of!(self is DatabricksDialect) => { + Keyword::STRUCT if dialect_of!(self is DatabricksDialect | RedshiftSqlDialect) => { + // Databricks and Redshift Spectrum (Hive-style) structs both + // use colon-separated `name:type` fields: + // `struct`. self.prev_token(); let (field_defs, _trailing_bracket) = self.parse_struct_type_def( Self::parse_databricks_query_struct_field_def, @@ -18197,7 +18681,9 @@ impl<'a> Parser<'a> { // Snowflake: EXECUTE TASK [ USING CONFIG = ] // https://docs.snowflake.com/en/sql-reference/sql/execute-task if self.parse_keyword(Keyword::TASK) { + let name_start = self.index; let name = self.parse_object_name(false)?; + let name = name.spanning(self.span_from_index(name_start)); if self.parse_keyword(Keyword::USING) { let _ = self.parse_identifier(false)?; let _ = self.consume_token(&Token::Eq); @@ -18951,7 +19437,7 @@ impl<'a> Parser<'a> { /// `LOAD [extension_name]` pub fn parse_load(&mut self) -> Result { let extension_name = self.parse_identifier(false)?; - Ok(Statement::Load { extension_name }) + Ok(Statement::DuckDbLoad { extension_name }) } /// ```sql @@ -19163,6 +19649,59 @@ impl<'a> Parser<'a> { Ok(NamedWindowDefinition(ident, named_window_expr)) } + /// Capture a routine / code body that is a single string literal — + /// dollar-quoted (`$$ ... $$` / `$tag$ ... $tag$`) or single-quoted + /// (`'...'`) — as a re-parseable [`FunctionDefinition`], preserving the + /// body's source start location so consumers can remap body-relative spans + /// back to the outer SQL. Returns `None` and consumes nothing if the next + /// token isn't such a string. Shared by `CREATE PROCEDURE` and the + /// standalone PostgreSQL `DO` block. + fn parse_function_body_definition(&mut self) -> Option { + let peek = self.peek_token(); + match peek.token.clone() { + Token::DollarQuotedString(s) => { + let body_start = dollar_quoted_body_start(peek.span.start, &s); + self.next_token(); + Some(FunctionDefinition::DoubleDollarDef { + value: s.value, + body_start, + }) + } + Token::SingleQuotedString(s) => { + // Opening `'` sits at peek.span.start; body follows it. + let body_start = Location { + line: peek.span.start.line, + column: peek.span.start.column.saturating_add(1), + }; + self.next_token(); + Some(FunctionDefinition::SingleQuotedDef { + value: s, + body_start, + }) + } + _ => None, + } + } + + /// Parse a PostgreSQL standalone anonymous code block: + /// `DO [ LANGUAGE ] [ LANGUAGE ]`. + /// + /// The leading `DO` has already been consumed. The `` string body is + /// captured as a re-parseable [`FunctionDefinition`] (typically dollar-quoted) + /// so lineage extraction can recover embedded DML/DDL; the `LANGUAGE` clause + /// (which may appear before or after the body) is consumed but not modelled. + /// + fn parse_do(&mut self) -> Result { + // Optional `LANGUAGE ` may precede the code block. + if self.parse_keyword(Keyword::LANGUAGE) { + let _ = self.parse_identifier(false)?; + } + let body = self.parse_function_body_definition(); + // Consume any trailing clause (e.g. a trailing `LANGUAGE `). + self.skip_to_statement_end(); + Ok(Statement::Do { body }) + } + pub fn parse_create_procedure(&mut self, or_alter: bool) -> Result { let name = self.parse_object_name(false)?; let params = self.parse_optional_procedure_parameters()?; @@ -19234,35 +19773,10 @@ impl<'a> Parser<'a> { // or another expression. Capture the raw definition string so // downstream consumers (lineage extractors) can re-parse it, then // skip the remainder of the statement. - let body_definition = { - let peek = self.peek_token(); - match peek.token.clone() { - Token::DollarQuotedString(s) => { - let body_start = dollar_quoted_body_start(peek.span.start, &s); - self.next_token(); - Some(FunctionDefinition::DoubleDollarDef { - value: s.value, - body_start, - }) - } - Token::SingleQuotedString(s) => { - // Opening `'` sits at peek.span.start; body follows it. - let body_start = Location { - line: peek.span.start.line, - column: peek.span.start.column.saturating_add(1), - }; - self.next_token(); - Some(FunctionDefinition::SingleQuotedDef { - value: s, - body_start, - }) - } - _ => { - let _ = self.parse_expr(); - None - } - } - }; + let body_definition = self.parse_function_body_definition(); + if body_definition.is_none() { + let _ = self.parse_expr(); + } // Consume trailing clauses like `LANGUAGE plpgsql`, `SECURITY INVOKER`, // `STABLE`, `VOLATILE`, etc. that follow the body in PostgreSQL/Redshift. loop { diff --git a/tests/sqlparser_bigquery.rs b/tests/sqlparser_bigquery.rs index ba6a5c998..7b88b11f7 100644 --- a/tests/sqlparser_bigquery.rs +++ b/tests/sqlparser_bigquery.rs @@ -2718,3 +2718,43 @@ fn parse_bigquery_at_time_zone_double_quoted_string() { .parse_sql_statements("SELECT ts AT TIME ZONE 'UTC'") .unwrap(); } + +#[test] +fn parse_create_model_as_select() { + // BigQuery ML: TRANSFORM/OPTIONS are dropped; the AS query (training input) + // is retained as real lineage. + let stmt = bigquery().one_statement_parses_to( + "CREATE MODEL `mydataset.mymodel` \ + TRANSFORM(f1 + f2 as c, * EXCEPT(f1, f2)) \ + OPTIONS(model_type='linear_reg', input_label_cols=['label_col']) \ + AS SELECT f1, f2, f3, label_col FROM t", + "CREATE MODEL `mydataset`.`mymodel` AS SELECT f1, f2, f3, label_col FROM t", + ); + match stmt { + Statement::CreateModel { + or_replace, + if_not_exists, + name, + query, + .. + } => { + assert!(!or_replace); + assert!(!if_not_exists); + assert_eq!(name.to_string(), "`mydataset`.`mymodel`"); + let query = query.expect("AS query captured"); + assert_eq!(query.to_string(), "SELECT f1, f2, f3, label_col FROM t"); + } + other => panic!("expected CreateModel, got {other:?}"), + } +} + +#[test] +fn parse_create_model_or_replace() { + match bigquery().one_statement_parses_to( + "CREATE OR REPLACE MODEL foo OPTIONS (model_type='linear_reg') AS SELECT bla FROM foo WHERE cond", + "CREATE OR REPLACE MODEL foo AS SELECT bla FROM foo WHERE cond", + ) { + Statement::CreateModel { or_replace, .. } => assert!(or_replace), + other => panic!("expected CreateModel, got {other:?}"), + } +} diff --git a/tests/sqlparser_common.rs b/tests/sqlparser_common.rs index 571d24b47..42d88112d 100644 --- a/tests/sqlparser_common.rs +++ b/tests/sqlparser_common.rs @@ -9250,3 +9250,93 @@ fn parse_teradata_column_attributes() { .unwrap_or_else(|e| panic!("failed to parse `{sql}`: {e}")); } } + +#[test] +fn parse_create_streaming_table_captures_query() { + // Databricks: CREATE OR REFRESH STREAMING TABLE ... AS . + // Routed through the table parser so the AS query is captured. + match generic().one_statement_parses_to( + "CREATE OR REFRESH STREAMING TABLE st AS SELECT * FROM stream_src", + "CREATE TABLE st AS SELECT * FROM stream_src", + ) { + Statement::CreateTable { name, query, .. } => { + assert_eq!(name.to_string(), "st"); + assert!(query.is_some(), "AS query should be captured"); + } + other => panic!("expected CreateTable, got {other:?}"), + } +} + +#[test] +fn parse_create_unsupported_distinct_from_comment() { + // Unmodelled CREATE statements yield CreateUnsupported, not a fake COMMENT. + match generic() + .parse_sql_statements("CREATE WAREHOUSE wh") + .unwrap() + .pop() + .unwrap() + { + Statement::CreateUnsupported { + or_replace, + object_type, + body, + } => { + assert!(!or_replace); + assert_eq!(object_type, "WAREHOUSE"); + assert_eq!(body, "wh"); + } + other => panic!("expected CreateUnsupported, got {other:?}"), + } +} + +#[test] +fn parse_drop_unsupported_captures_body() { + // Unmodelled DROP object types are captured opaquely instead of erroring. + match generic() + .parse_sql_statements("DROP WAREHOUSE wh") + .unwrap() + .pop() + .unwrap() + { + Statement::DropUnsupported { object_type, body } => { + assert_eq!(object_type, "WAREHOUSE"); + assert_eq!(body, "wh"); + } + other => panic!("expected DropUnsupported, got {other:?}"), + } +} + +#[test] +fn parse_alter_unsupported_captures_body() { + match generic() + .parse_sql_statements("ALTER WAREHOUSE wh SET WAREHOUSE_SIZE = 'LARGE'") + .unwrap() + .pop() + .unwrap() + { + Statement::AlterUnsupported { object_type, body } => { + assert_eq!(object_type, "WAREHOUSE"); + assert!(body.contains("WAREHOUSE_SIZE")); + } + other => panic!("expected AlterUnsupported, got {other:?}"), + } +} + +#[test] +fn parse_unsupported_statement_not_masquerading_as_set() { + // ClickHouse SYSTEM / Postgres LOCK / DO are captured honestly, not as a + // fake SET statement. The body text is preserved. + match generic() + .parse_sql_statements("LOCK TABLE foo IN ACCESS EXCLUSIVE MODE") + .unwrap() + .pop() + .unwrap() + { + Statement::UnsupportedStatement { keyword, body } => { + assert_eq!(keyword, "LOCK"); + // The locked table reference survives in the captured body. + assert!(body.contains("foo")); + } + other => panic!("expected UnsupportedStatement, got {other:?}"), + } +} diff --git a/tests/sqlparser_databricks.rs b/tests/sqlparser_databricks.rs index 2b84d68f0..d13a836c3 100644 --- a/tests/sqlparser_databricks.rs +++ b/tests/sqlparser_databricks.rs @@ -625,3 +625,40 @@ fn test_declare_variable() { .unwrap_or_else(|e| panic!("failed to parse {sql:?}: {e}")); } } + +#[test] +fn parse_create_materialized_view_skips_schedule_and_trigger() { + // Databricks MV pre-AS clauses (SCHEDULE / TRIGGER) carry no lineage and are + // skipped; the AS query (the lineage) must be preserved. + for (sql, canonical) in [ + ( + "CREATE MATERIALIZED VIEW daily_sales SCHEDULE EVERY 1 DAY \ + AS SELECT d, SUM(sales) AS s FROM table1 GROUP BY d", + "CREATE MATERIALIZED VIEW daily_sales \ + AS SELECT d, SUM(sales) AS s FROM table1 GROUP BY d", + ), + ( + "CREATE MATERIALIZED VIEW dr SCHEDULE CRON '0 0 0 * * ?' AT TIME ZONE 'UTC' \ + AS SELECT d FROM orders", + "CREATE MATERIALIZED VIEW dr AS SELECT d FROM orders", + ), + ( + "CREATE MATERIALIZED VIEW IF NOT EXISTS sm TRIGGER ON UPDATE \ + AS SELECT a FROM movies", + "CREATE MATERIALIZED VIEW IF NOT EXISTS sm AS SELECT a FROM movies", + ), + ] { + match databricks().one_statement_parses_to(sql, canonical) { + Statement::CreateView { + materialized, + query, + .. + } => { + assert!(materialized); + // The upstream table reference survives in the query body. + assert!(query.to_string().contains("FROM")); + } + other => panic!("expected materialized CreateView, got {other:?}"), + } + } +} diff --git a/tests/sqlparser_hive.rs b/tests/sqlparser_hive.rs index 7ce998a95..68b6caefd 100644 --- a/tests/sqlparser_hive.rs +++ b/tests/sqlparser_hive.rs @@ -568,3 +568,43 @@ fn parse_hive_create_table_comment_and_clustered_by() { .unwrap_or_else(|e| panic!("failed to parse `{sql}`: {e}")); } } + +#[test] +fn parse_load_data_inpath_into_table() { + // Hive/Spark: LOAD DATA [LOCAL] INPATH '' [OVERWRITE] INTO TABLE . + // The target table (output) and the inpath (file source) are captured. + match hive().verified_stmt("LOAD DATA INPATH '/tmp/x.csv' INTO TABLE mytable") { + Statement::LoadData { + local, + inpath, + overwrite, + target, + .. + } => { + assert!(!local); + assert_eq!(inpath.as_deref(), Some("/tmp/x.csv")); + assert!(!overwrite); + assert_eq!(target, ObjectName(vec!["mytable".into()])); + } + other => panic!("expected LoadData, got {other:?}"), + } + // LOCAL + OVERWRITE + PARTITION (the partition list is opaque `rest`). + match hive().verified_stmt( + "LOAD DATA LOCAL INPATH '/d/c2=2' OVERWRITE INTO TABLE t PARTITION ( c2 = 2 )", + ) { + Statement::LoadData { + local, + inpath, + overwrite, + target, + rest, + } => { + assert!(local); + assert_eq!(inpath.as_deref(), Some("/d/c2=2")); + assert!(overwrite); + assert_eq!(target.to_string(), "t"); + assert!(rest.contains("PARTITION")); + } + other => panic!("expected LoadData, got {other:?}"), + } +} diff --git a/tests/sqlparser_postgres.rs b/tests/sqlparser_postgres.rs index e78bd96a6..c41b551b6 100644 --- a/tests/sqlparser_postgres.rs +++ b/tests/sqlparser_postgres.rs @@ -4660,3 +4660,86 @@ fn parse_create_table_as_with_data_clause() { .unwrap_or_else(|e| panic!("failed to parse `{sql}`: {e}")); } } + +#[test] +fn parse_create_foreign_table_captures_columns() { + // PostgreSQL foreign table: routed through external-table handling so the + // name/columns (lineage schema) are captured; SERVER/OPTIONS are skipped. + let stmt = pg() + .parse_sql_statements( + "CREATE FOREIGN TABLE words (word text NOT NULL) SERVER local_file OPTIONS (filename '/usr/share/dict/words')", + ) + .unwrap() + .pop() + .unwrap(); + assert_eq!( + stmt.to_string(), + "CREATE EXTERNAL TABLE words (word TEXT NOT NULL)" + ); + match stmt { + Statement::CreateTable { + name, + columns, + external, + .. + } => { + assert!(external); + assert_eq!(name.to_string(), "words"); + assert_eq!(columns.len(), 1); + assert_eq!(columns[0].name.to_string(), "word"); + } + other => panic!("expected external CreateTable, got {other:?}"), + } +} + +#[test] +fn parse_do_block_captures_reparseable_body() { + // PostgreSQL standalone anonymous code block: the $$...$$ body is captured + // as a re-parseable FunctionDefinition (not opaque text), so lineage + // extraction can recover embedded DML. + let sql = "DO $$ BEGIN INSERT INTO t SELECT a FROM src; END $$"; + match pg().verified_stmt(sql) { + Statement::Do { body } => { + let body = body.expect("DO body captured"); + match body { + FunctionDefinition::DoubleDollarDef { value, .. } => { + // The embedded DML is recoverable from the captured body. + assert!(value.contains("INSERT INTO t")); + } + other => panic!("expected dollar-quoted body, got {other:?}"), + } + } + other => panic!("expected Do, got {other:?}"), + } + // LANGUAGE clause (before or after the body) is consumed. + pg().one_statement_parses_to( + "DO LANGUAGE plpgsql $$ BEGIN PERFORM 1; END $$", + "DO $$ BEGIN PERFORM 1; END $$", + ); +} + +#[test] +fn parse_do_records_body_start() { + // The DO body must record body_start identically to CREATE PROCEDURE so + // consumers can remap body-relative spans. "DO " is 3 chars, so the opening + // `$$` is at column 4 and the body's first char is at column 6. + use sqlparser::parser::Parser; + use sqlparser::tokenizer::{Location, Tokenizer}; + let sql = "DO $$ BEGIN NULL; END $$"; + let dialect = PostgreSqlDialect {}; + let tokens = Tokenizer::new(&dialect, sql) + .tokenize_with_location() + .unwrap(); + let mut stmts = Parser::new(&dialect) + .with_tokens_with_locations(tokens) + .parse_statements() + .unwrap(); + match stmts.pop().unwrap() { + Statement::Do { + body: Some(FunctionDefinition::DoubleDollarDef { body_start, .. }), + } => { + assert_eq!(body_start, Location { line: 1, column: 6 }); + } + other => panic!("expected Do with dollar body, got {other:?}"), + } +} diff --git a/tests/sqlparser_redshift.rs b/tests/sqlparser_redshift.rs index 0e593e2a9..d79f54afe 100644 --- a/tests/sqlparser_redshift.rs +++ b/tests/sqlparser_redshift.rs @@ -932,3 +932,157 @@ fn parse_redshift_identity_seed_step() { ) .unwrap(); } + +#[test] +fn parse_create_external_table_spectrum_nested_types() { + // Redshift Spectrum (Hive-style) nested column types: struct<>/array<>/map<>. + // Column names and element types must be preserved for lineage. + let stmt = redshift().verified_stmt( + "CREATE EXTERNAL TABLE spectrum.customers (id INT, \ + name STRUCT, \ + phones ARRAY, \ + orders ARRAY>) \ + STORED AS PARQUET LOCATION 's3://bucket/customers/'", + ); + match stmt { + Statement::CreateTable { + name, + columns, + external, + .. + } => { + assert!(external); + assert_eq!(name.to_string(), "spectrum.customers"); + assert_eq!(columns.len(), 4); + // The struct column keeps its named, typed fields. + match &columns[1].data_type { + DataType::Struct(fields) => { + assert_eq!(fields.len(), 2); + assert_eq!(fields[0].field_name.as_ref().unwrap().to_string(), "given"); + } + other => panic!("expected STRUCT, got {other:?}"), + } + assert!(matches!(columns[2].data_type, DataType::Array(_))); + } + other => panic!("expected external CreateTable, got {other:?}"), + } +} + +#[test] +fn parse_create_external_table_spectrum_map_type() { + let stmt = redshift().verified_stmt( + "CREATE EXTERNAL TABLE spectrum.t (phones MAP) \ + STORED AS PARQUET LOCATION 's3://bucket/t/'", + ); + match stmt { + Statement::CreateTable { columns, .. } => { + assert!(matches!(columns[0].data_type, DataType::DatabricksMap(_))); + } + other => panic!("expected external CreateTable, got {other:?}"), + } +} + +#[test] +fn parse_create_external_schema_data_catalog() { + let stmt = redshift().one_statement_parses_to( + "create external schema spectrum_schema from data catalog \ + database 'sampledb' region 'us-west-2' \ + iam_role 'arn:aws:iam::123456789012:role/MySpectrumRole'", + "CREATE EXTERNAL SCHEMA spectrum_schema FROM DATA CATALOG DATABASE 'sampledb'", + ); + match stmt { + Statement::CreateExternalSchema { + if_not_exists, + name, + from, + database, + schema, + } => { + assert!(!if_not_exists); + assert_eq!(name.to_string(), "spectrum_schema"); + assert_eq!(from, "DATA CATALOG"); + assert_eq!(database.as_deref(), Some("sampledb")); + assert_eq!(schema, None); + } + other => panic!("expected CreateExternalSchema, got {other:?}"), + } +} + +#[test] +fn parse_create_external_schema_hive_and_redshift() { + match redshift().one_statement_parses_to( + "create external schema hive_schema from hive metastore \ + database 'hive_db' uri '172.10.10.10' port 99 iam_role 'arn:...'", + "CREATE EXTERNAL SCHEMA hive_schema FROM HIVE METASTORE DATABASE 'hive_db'", + ) { + Statement::CreateExternalSchema { from, database, .. } => { + assert_eq!(from, "HIVE METASTORE"); + assert_eq!(database.as_deref(), Some("hive_db")); + } + other => panic!("expected CreateExternalSchema, got {other:?}"), + } + // Federated Redshift source carries both DATABASE and SCHEMA. + match redshift().verified_stmt( + "CREATE EXTERNAL SCHEMA sales_schema FROM REDSHIFT DATABASE 'sales_db' SCHEMA 'public'", + ) { + Statement::CreateExternalSchema { + from, + database, + schema, + .. + } => { + assert_eq!(from, "REDSHIFT"); + assert_eq!(database.as_deref(), Some("sales_db")); + assert_eq!(schema.as_deref(), Some("public")); + } + other => panic!("expected CreateExternalSchema, got {other:?}"), + } +} + +#[test] +fn parse_create_model_from_subquery() { + // Redshift ML: FROM () is the training input — the subquery's + // FROM/WHERE/columns must survive for lineage. + let stmt = redshift().one_statement_parses_to( + "CREATE MODEL customer_churn_auto_model \ + FROM (SELECT state, churn FROM customer_activity WHERE record_date < '2020-01-01') \ + TARGET churn FUNCTION ml_fn_customer_churn_auto \ + IAM_ROLE default SETTINGS (S3_BUCKET 'amzn-s3-demo-bucket')", + "CREATE MODEL customer_churn_auto_model AS SELECT state, churn \ + FROM customer_activity WHERE record_date < '2020-01-01'", + ); + match stmt { + Statement::CreateModel { name, query, .. } => { + assert_eq!(name.to_string(), "customer_churn_auto_model"); + let query = query.expect("FROM (subquery) captured as training input"); + assert!(query.to_string().contains("FROM customer_activity")); + } + other => panic!("expected CreateModel, got {other:?}"), + } +} + +#[test] +fn parse_create_model_from_bare_table() { + // Redshift ML: FROM
(not a subquery) — the source table must be + // recorded so the training-input lineage edge is captured. + let stmt = redshift().one_statement_parses_to( + "CREATE MODEL churn_model FROM customer_activity \ + TARGET churn FUNCTION predict_churn IAM_ROLE default \ + SETTINGS (S3_BUCKET 'my-bucket')", + "CREATE MODEL churn_model FROM customer_activity", + ); + match stmt { + Statement::CreateModel { + name, + query, + from_table, + .. + } => { + assert_eq!(name.to_string(), "churn_model"); + assert!(query.is_none()); + let from_table = from_table.expect("FROM
source captured"); + assert_eq!(from_table.to_string(), "customer_activity"); + } + other => panic!("expected CreateModel, got {other:?}"), + } +} diff --git a/tests/sqlparser_snowflake.rs b/tests/sqlparser_snowflake.rs index 5c0c9aa40..89267d421 100644 --- a/tests/sqlparser_snowflake.rs +++ b/tests/sqlparser_snowflake.rs @@ -4293,3 +4293,312 @@ fn parse_snowflake_table_function_tablesample() { .parse_sql_statements("SELECT * FROM TABLE('t1') TABLESAMPLE BERNOULLI (20.3)") .unwrap(); } + +/// Helper: parse a single statement with location tracking enabled so the +/// captured `WithSpan` location is meaningful (the convenience `parse_sql` +/// path used by `verified_stmt` installs dummy spans). +fn snowflake_parse_with_locations(sql: &str) -> Statement { + use sqlparser::parser::Parser; + use sqlparser::tokenizer::Tokenizer; + let dialect = SnowflakeDialect {}; + let tokens = Tokenizer::new(&dialect, sql) + .tokenize_with_location() + .unwrap(); + let mut stmts = Parser::new(&dialect) + .with_tokens_with_locations(tokens) + .parse_statements() + .unwrap(); + assert_eq!(stmts.len(), 1); + stmts.pop().unwrap() +} + +#[test] +fn parse_create_task_with_insert_body() { + // Scheduling/warehouse properties are dropped; the AS body is retained. + let stmt = snowflake().one_statement_parses_to( + "CREATE TASK mytask_minute \ + WAREHOUSE = mywh \ + SCHEDULE = '5 MINUTES' \ + AS INSERT INTO mytable(ts) VALUES(CURRENT_TIMESTAMP)", + "CREATE TASK mytask_minute AS INSERT INTO mytable (ts) VALUES (CURRENT_TIMESTAMP)", + ); + match stmt { + Statement::CreateTask { + or_replace, + or_alter, + if_not_exists, + name, + after, + when, + body, + } => { + assert!(!or_replace); + assert!(!or_alter); + assert!(!if_not_exists); + assert_eq!(name.to_string(), "mytask_minute"); + assert!(after.is_empty()); + assert!(when.is_none()); + // The body is the real INSERT statement (lineage preserved). + assert!(matches!(body.as_deref(), Some(Statement::Insert { .. }))); + } + other => panic!("expected CreateTask, got {other:?}"), + } +} + +#[test] +fn parse_create_task_with_after_dependencies() { + let stmt = snowflake().verified_stmt( + "CREATE TASK task5 AFTER task2, task3, task4 AS INSERT INTO t1 (ts) VALUES (CURRENT_TIMESTAMP)", + ); + match stmt { + Statement::CreateTask { name, after, .. } => { + assert_eq!(name.to_string(), "task5"); + let deps: Vec = after.iter().map(|d| d.to_string()).collect(); + assert_eq!(deps, vec!["task2", "task3", "task4"]); + } + other => panic!("expected CreateTask, got {other:?}"), + } +} + +#[test] +fn parse_create_task_with_when_predicate() { + let stmt = snowflake().one_statement_parses_to( + "CREATE TASK triggered_task_stream \ + WHEN SYSTEM$STREAM_HAS_DATA('orders_stream') \ + AS INSERT INTO completed_promotions SELECT order_id FROM orders_stream", + "CREATE TASK triggered_task_stream WHEN SYSTEM$STREAM_HAS_DATA('orders_stream') \ + AS INSERT INTO completed_promotions SELECT order_id FROM orders_stream", + ); + match stmt { + Statement::CreateTask { when, body, .. } => { + let when = when.expect("WHEN predicate captured"); + assert_eq!(when.to_string(), "SYSTEM$STREAM_HAS_DATA('orders_stream')"); + assert!(matches!(body.as_deref(), Some(Statement::Insert { .. }))); + } + other => panic!("expected CreateTask, got {other:?}"), + } +} + +#[test] +fn parse_create_task_or_replace_and_or_alter() { + match snowflake().verified_stmt("CREATE OR REPLACE TASK t AS SELECT 1") { + Statement::CreateTask { + or_replace, + or_alter, + .. + } => { + assert!(or_replace); + assert!(!or_alter); + } + other => panic!("expected CreateTask, got {other:?}"), + } + // OR ALTER + dotted/qualified name + EXECUTE AS USER (the EXECUTE AS pair + // must not be mistaken for the AS body marker). + match snowflake().one_statement_parses_to( + "CREATE OR ALTER TASK team_task SCHEDULE='12 HOURS' EXECUTE AS USER task_user AS SELECT 1", + "CREATE OR ALTER TASK team_task AS SELECT 1", + ) { + Statement::CreateTask { + or_replace, + or_alter, + name, + .. + } => { + assert!(!or_replace); + assert!(or_alter); + assert_eq!(name.to_string(), "team_task"); + } + other => panic!("expected CreateTask, got {other:?}"), + } +} + +#[test] +fn parse_create_task_begin_block_body() { + // Snowflake scripting BEGIN..END block as the task body. + match snowflake().parse_sql_statements( + "CREATE OR REPLACE TASK test_logging \ + SCHEDULE = 'USING CRON 0 * * * * America/Los_Angeles' \ + AS BEGIN SELECT CURRENT_TIMESTAMP; END", + ) { + Ok(stmts) => match &stmts[0] { + Statement::CreateTask { body, .. } => { + assert!(body.is_some(), "BEGIN..END block body should be captured"); + } + other => panic!("expected CreateTask, got {other:?}"), + }, + Err(e) => panic!("parse failed: {e}"), + } +} + +#[test] +fn parse_create_task_unparseable_body_falls_back() { + // `EXECUTE DBT PROJECT` is not modelled; the body must degrade to None + // rather than corrupting the surrounding statement. + match snowflake().parse_sql_statements( + "CREATE OR ALTER TASK run_dbt WAREHOUSE = wh \ + AS EXECUTE DBT PROJECT my_db.my_schema.my_proj args='run --target prod'", + ) { + Ok(stmts) => match &stmts[0] { + Statement::CreateTask { name, body, .. } => { + assert_eq!(name.to_string(), "run_dbt"); + assert!(body.is_none(), "unparseable body should be None"); + } + other => panic!("expected CreateTask, got {other:?}"), + }, + Err(e) => panic!("parse should not error: {e}"), + } +} + +#[test] +fn parse_create_task_captures_name_location() { + // Verify the name span is populated when location tracking is on. + let stmt = snowflake_parse_with_locations("CREATE TASK mytask AS SELECT 1"); + match stmt { + Statement::CreateTask { name, .. } => { + let span = name.span_location(); + assert_ne!( + span, + Span::default(), + "task name location should be captured" + ); + // "CREATE TASK " is 12 chars, so the name starts at column 13. + assert_eq!(span.start.column, 13); + } + other => panic!("expected CreateTask, got {other:?}"), + } +} + +#[test] +fn parse_create_stream_on_table_view_stage() { + for (sql, src_type, src) in [ + ( + "CREATE STREAM mystream ON TABLE mytable", + "TABLE", + "mytable", + ), + ("CREATE STREAM mystream ON VIEW myview", "VIEW", "myview"), + ("CREATE STREAM s ON STAGE mystage", "STAGE", "mystage"), + ] { + match snowflake().verified_stmt(sql) { + Statement::CreateStream { + source_type, + source, + .. + } => { + assert_eq!(source_type, src_type); + assert_eq!(source.to_string(), src); + } + other => panic!("expected CreateStream, got {other:?}"), + } + } +} + +#[test] +fn parse_create_stream_skips_time_travel() { + // AT/BEFORE time-travel and APPEND_ONLY etc. are not lineage and are dropped. + match snowflake().one_statement_parses_to( + "CREATE OR REPLACE STREAM mystream ON TABLE mytable AT(STREAM => 'mystream')", + "CREATE OR REPLACE STREAM mystream ON TABLE mytable", + ) { + Statement::CreateStream { + or_replace, + name, + source_type, + source, + .. + } => { + assert!(or_replace); + assert_eq!(name.to_string(), "mystream"); + assert_eq!(source_type, "TABLE"); + assert_eq!(source.to_string(), "mytable"); + } + other => panic!("expected CreateStream, got {other:?}"), + } +} + +#[test] +fn parse_create_stream_external_table_source() { + match snowflake().verified_stmt("CREATE STREAM s ON EXTERNAL TABLE ext.src") { + Statement::CreateStream { + source_type, + source, + .. + } => { + assert_eq!(source_type, "EXTERNAL TABLE"); + assert_eq!(source.to_string(), "ext.src"); + } + other => panic!("expected CreateStream, got {other:?}"), + } +} + +#[test] +fn parse_create_pipe_copy_into_body() { + let stmt = snowflake().one_statement_parses_to( + "CREATE PIPE mypipe_aws AUTO_INGEST = TRUE \ + AS COPY INTO snowpipe_db.public.mytable FROM @snowpipe_db.public.mystage \ + FILE_FORMAT = (TYPE = 'JSON')", + "CREATE PIPE mypipe_aws AS COPY INTO snowpipe_db.public.mytable \ + FROM @snowpipe_db.public.mystage FILE_FORMAT=(TYPE='JSON')", + ); + match stmt { + Statement::CreatePipe { name, body, .. } => { + assert_eq!(name.to_string(), "mypipe_aws"); + // The body is the real COPY INTO statement carrying ingest lineage. + assert!(matches!( + body.as_deref(), + Some(Statement::CopyIntoSnowflake { .. }) + )); + } + other => panic!("expected CreatePipe, got {other:?}"), + } +} + +#[test] +fn parse_execute_task_captures_name_location() { + match snowflake().verified_stmt("EXECUTE TASK mytask") { + Statement::ExecuteTask { name } => assert_eq!(name.to_string(), "mytask"), + other => panic!("expected ExecuteTask, got {other:?}"), + } + // USING CONFIG is consumed but not represented. + snowflake().one_statement_parses_to( + "EXECUTE TASK my_root_task USING CONFIG=$${\"a\": 1}$$", + "EXECUTE TASK my_root_task", + ); + // Location is captured under location tracking. + match snowflake_parse_with_locations("EXECUTE TASK mytask") { + Statement::ExecuteTask { name } => { + assert_ne!(name.span_location(), Span::default()); + } + other => panic!("expected ExecuteTask, got {other:?}"), + } +} + +#[test] +fn parse_create_unsupported_is_not_a_comment() { + // CREATE WAREHOUSE is not modelled — it must produce CreateUnsupported, + // NOT a fake COMMENT statement. + match snowflake().parse_sql_statements("CREATE WAREHOUSE wh WITH WAREHOUSE_SIZE='XSMALL'") { + Ok(stmts) => match &stmts[0] { + Statement::CreateUnsupported { + or_replace, + object_type, + body, + } => { + assert!(!or_replace); + assert_eq!(object_type, "WAREHOUSE"); + // The remaining text is captured verbatim. + assert!(body.contains("wh")); + assert!(body.contains("WAREHOUSE_SIZE")); + } + other => panic!("expected CreateUnsupported, got {other:?}"), + }, + Err(e) => panic!("parse failed: {e}"), + } + match snowflake().parse_sql_statements("CREATE OR REPLACE FILE FORMAT ff TYPE='CSV'") { + Ok(stmts) => match &stmts[0] { + Statement::CreateUnsupported { or_replace, .. } => assert!(or_replace), + other => panic!("expected CreateUnsupported, got {other:?}"), + }, + Err(e) => panic!("parse failed: {e}"), + } +}