swellow 0.2.0

SQL-first migration CLI
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207

mod dialect;
mod error;
mod io;
pub mod migration;
mod resource;
mod statement;

pub use dialect::ReferenceToStaticDialect;
pub use error::{ParseError, ParseErrorKind};
pub use io::collect_versions_from_directory;
pub use resource::{Resource, ResourceCollection};
pub use statement::StatementCollection;

use sqlparser::ast::Statement;
use sqlparser::parser::Parser;
use sqlparser::tokenizer::Token;


/// Parse as many tokens as possible into a statement.
/// This is a bandaid to any limitation in the parser's lexicon.
/// If a statement is valid, but not yet in the parser's lexicon, it would fail.
/// This shouldn't be the case.
/// After all, parsing is only required up to the subject of the statement,
/// e.g. "CREATE TABLE table_name" - in order to track the resource being changed.
/// Everything after is irrelevant.
pub fn greedy_parse(
    dialect: ReferenceToStaticDialect,
    tokens: Vec<Token>
) -> Option<Statement> {
    let mut last_ok = None;

    for i in 1..=tokens.len() {
        let partial = &tokens[..i];

        match Parser::new(dialect)
            .with_tokens(partial.to_vec())
            .parse_statements() {
                Ok(stmt) => last_ok = stmt.first().cloned(),
                Err(e) => tracing::debug!(
                    "SQL parsing failed at token {}: {:?} ({})",
                    i,
                    tokens.get(i - 1), // Use .get() to avoid panic on empty tokens
                    e
                )
            };
    }

    last_ok
}


/// This test suite has a number of aims:
/// 1) Test that the collections are successfully initialized.
/// 2) Test that most SQL statements will be, to a variable extent, parsed successfully.
/// 3) Test that some SQL statements are outright impossible to parse.
/// 4) Test that regardless of how a statement is parsed, the original query can be restored from a vector of tokens.
/// 5) Test that statements that are successfully parsed can be parsed into a meaningful resource collection.
#[cfg(test)]
mod tests {
    use crate::parser::{dialect::*, ResourceCollection, StatementCollection};

    fn make_collection(dialect: ReferenceToStaticDialect, sql: &str) -> StatementCollection {
        StatementCollection::new(dialect)
            .parse_sql(sql).unwrap()
    }

    #[test]
    fn test_parsed_some_part() {
        let valid_cases: &[(ReferenceToStaticDialect, &str, &str)] = &[
            // All dialects
            (&DIALECT_POSTGRES, "SELECT a,, b FROM t;", ""),
            (&DIALECT_HIVE, "SELECT a,, b FROM t;", ""),
            (&DIALECT_DATABRICKS, "SELECT a,, b FROM t;", ""),
            (&DIALECT_POSTGRES, "SELECT a, b, FROM t;", ""),
            (&DIALECT_HIVE, "SELECT a, b, FROM t;", ""),
            (&DIALECT_DATABRICKS, "SELECT a, b, FROM t;", ""),
            (&DIALECT_POSTGRES, "SELECT * FROM my_table WHERE;", ""),
            (&DIALECT_HIVE, "SELECT * FROM my_table WHERE;", ""),
            (&DIALECT_DATABRICKS, "SELECT * FROM my_table WHERE;", ""),
            (&DIALECT_POSTGRES, "SELECT * FROM my_table WHERE id =;", ""),
            (&DIALECT_HIVE, "SELECT * FROM my_table WHERE id =;", ""),
            (&DIALECT_DATABRICKS, "SELECT * FROM my_table WHERE id =;", ""),
            (&DIALECT_POSTGRES, "SELECT my_col;", ""),
            (&DIALECT_HIVE, "SELECT my_col;", ""),
            (&DIALECT_DATABRICKS, "SELECT my_col;", ""),
            (&DIALECT_POSTGRES, "SELECT * FROM a LEFT JOIN b;", ""),
            (&DIALECT_HIVE, "SELECT * FROM a LEFT JOIN b;", ""),
            (&DIALECT_DATABRICKS, "SELECT * FROM a LEFT JOIN b;", ""),
            (&DIALECT_POSTGRES, "SELECT * FROM a JOIN b USING ();", ""),
            (&DIALECT_HIVE, "SELECT * FROM a JOIN b USING ();", ""),
            (&DIALECT_DATABRICKS, "SELECT * FROM a JOIN b USING ();", ""),
            
            (&DIALECT_POSTGRES, "SELECT * FROM t ORDER BY;", ""),
            (&DIALECT_HIVE, "SELECT * FROM t ORDER BY;", ""),
            (&DIALECT_DATABRICKS, "SELECT * FROM t ORDER BY;", ""),
            (&DIALECT_POSTGRES, "SELECT * FROM t GROUP BY;", ""),
            (&DIALECT_HIVE, "SELECT * FROM t GROUP BY;", ""),
            (&DIALECT_DATABRICKS, "SELECT * FROM t GROUP BY;", ""),
            
            // Postgres only
            (&DIALECT_POSTGRES, "SELECT DISTINCT FROM t;", ""),
            (&DIALECT_POSTGRES, "SELECT TOP 10 FROM t;", ""),
            (&DIALECT_POSTGRES, "SELECT ARRAY[1,2,3];", ""),
            (&DIALECT_POSTGRES, "SELECT * FROM my_table LIMIT;", ""),
            (&DIALECT_POSTGRES, "INSERT INTO my_table (id, name) VALUES (1, 'Alice') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name;", ""),
            
            // Databricks / Delta only
            (&DIALECT_DATABRICKS, "CREATE TABLE delta_table (id INT, value STRING) USING delta TBLPROPERTIES ('delta.autoOptimize.optimizeWrite' = true);", "CREATE TABLE"),
            (&DIALECT_DATABRICKS, "MERGE INTO target t USING source s ON t.id = s.id WHEN MATCHED THEN UPDATE SET t.value = s.value WHEN NOT MATCHED THEN INSERT (id, value) VALUES (s.id, s.value);", ""),
            (&DIALECT_DATABRICKS, "MERGE INTO target t USING source s ON t.id = s.id WHEN MATCHED THEN DELETE WHEN MATCHED AND s.value > 10 THEN UPDATE SET t.value = s.value;", ""),
            (&DIALECT_DATABRICKS, "CREATE TABLE t3 (id INT) USING delta PARTITIONED BY ();", "CREATE TABLE"),

            // Hive / Iceberg only
            (&DIALECT_HIVE, "CREATE TABLE iceberg_table (id INT, value STRING) USING iceberg PARTITIONED BY (days(ts));", "CREATE TABLE"),
            (&DIALECT_HIVE, "ALTER TABLE iceberg_table DROP COLUMN extra;", "ALTER TABLE"),
        ];
        
        for (dialect, sql, operation) in valid_cases {
            let collection = make_collection(*dialect, sql);
            assert!(!collection.is_empty(), "Failed parsing for dialect {:?}: {:?}", dialect, sql);

            assert!(sql == &collection.to_string());

            let resources = ResourceCollection::from_statement_collection(&collection);

            for resource in resources.iter() {
                // We take the first statement, because these queries are single-statement.
                assert!(operation == &format!("{} {}", resource.statements.first().unwrap(), resource.object_type))
            }
        }
    }

    #[test]
    fn test_failed_to_parse_any_part() {
        let invalid_cases: &[(ReferenceToStaticDialect, &str)] = &[
            // Postgres only
            (&DIALECT_POSTGRES, "INSERT INTO my_table (id, name) VALUES ();"),
            (&DIALECT_POSTGRES, "UPDATE my_table SET name =;"),

            // Databricks only
            (&DIALECT_DATABRICKS, "SELECT TOP 10 FROM t;"),
            (&DIALECT_DATABRICKS, "ALTER TABLE delta_table ADD COLUMNS (extra STRING);"),
            (&DIALECT_DATABRICKS, "SELECT DISTINCT FROM t;"),

            // Hive only
            (&DIALECT_HIVE, "SELECT TOP 10 FROM t;"),
            (&DIALECT_HIVE, "SELECT DISTINCT FROM t;"),
        ];
        
        for (dialect, sql) in invalid_cases {
            let collection = make_collection(*dialect, sql);
            assert!(collection.is_empty(), "Parsing should have failed for dialect {:?}: {:?}", dialect, sql);

            for s in collection.to_strings() {
                assert!(&s == sql)
            }
        }
    }

    #[test]
    fn test_multi_statement_queries() {
        let cases: &[(ReferenceToStaticDialect, &str)] = &[(
            &DIALECT_POSTGRES,
            "CREATE TABLE my_table (id INT, name TEXT); INSERT INTO my_table VALUES (1, 'Alice'); SELECT * FROM my_table;",
        ), (
            &DIALECT_DATABRICKS,
            "CREATE TABLE delta_table (id INT, name STRING) USING delta; INSERT INTO delta_table VALUES (1, 'Bob'); MERGE INTO delta_table AS t USING updates AS s ON t.id = s.id WHEN MATCHED THEN UPDATE SET t.name = s.name;",
        ), (
            &DIALECT_HIVE,
            "CREATE TABLE iceberg_table (id INT, value STRING) USING iceberg; ALTER TABLE iceberg_table ADD COLUMN new_col STRING; SELECT * FROM iceberg_table;",
        )];

        for (dialect, sql) in cases {
            let collection = make_collection(*dialect, sql);
            assert!(
                !collection.is_empty(),
                "Failed to parse multi-statement query for dialect {:?}: {:?}",
                dialect,
                sql
            );

            let statements = collection.to_strings();
            assert!(
                statements.len() > 1,
                "Expected multiple statements, got {} for dialect {:?}",
                statements.len(),
                dialect
            );

            // Make sure each statement round-trips cleanly
            for s in &statements {
                assert!(!s.trim().is_empty(), "Parsed empty statement for {:?}", dialect);
            }
            // Assert .to_string() maintains multi-statement integrity
            assert!(sql == &collection.to_string());

            // Ensure we can collect resources for each statement
            let resources = ResourceCollection::from_statement_collection(&collection);
            assert!(
                !resources.is_empty(),
                "No resources parsed from multi-statement query for {:?}",
                dialect
            );
        }
    }

}