risingwave_sqlparser/ast/
mod.rs

1// Licensed under the Apache License, Version 2.0 (the "License");
2// you may not use this file except in compliance with the License.
3// You may obtain a copy of the License at
4//
5//     http://www.apache.org/licenses/LICENSE-2.0
6//
7// Unless required by applicable law or agreed to in writing, software
8// distributed under the License is distributed on an "AS IS" BASIS,
9// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
10// See the License for the specific language governing permissions and
11// limitations under the License.
12
13//! SQL Abstract Syntax Tree (AST) types
14mod analyze;
15mod data_type;
16pub(crate) mod ddl;
17mod legacy_source;
18mod operator;
19mod query;
20mod statement;
21mod value;
22
23use std::collections::HashSet;
24use std::fmt::{self, Display};
25use std::sync::Arc;
26
27use itertools::Itertools;
28use winnow::ModalResult;
29
30pub use self::data_type::{DataType, StructField};
31pub use self::ddl::{
32    AlterColumnOperation, AlterCompactionGroupOperation, AlterConnectionOperation,
33    AlterDatabaseOperation, AlterFragmentOperation, AlterFunctionOperation, AlterRateLimit,
34    AlterRateLimitType, AlterSchemaOperation, AlterSecretOperation, AlterTableOperation, ColumnDef,
35    ColumnOption, ColumnOptionDef, ReferentialAction, SourceWatermark, TableConstraint,
36    WebhookSourceInfo,
37};
38pub use self::legacy_source::{CompatibleFormatEncode, get_delimiter};
39pub use self::operator::{BinaryOperator, QualifiedOperator, UnaryOperator};
40pub use self::query::{
41    AfterMatchSkip, Corresponding, Cte, CteInner, Distinct, Fetch, Join, JoinConstraint,
42    JoinOperator, LateralView, MatchRecognizePattern, MatchRecognizeSymbol, Measure, NamedWindow,
43    OrderByExpr, Query, RepetitionQuantifier, RowsPerMatch, Select, SelectItem, SetExpr,
44    SetOperator, SubsetDefinition, SymbolDefinition, TableAlias, TableFactor, TableWithJoins, Top,
45    Values, With,
46};
47pub use self::statement::*;
48pub use self::value::{
49    ConnectionRefValue, CstyleEscapedString, DateTimeField, DollarQuotedString, JsonPredicateType,
50    SecretRefAsType, SecretRefValue, TrimWhereField, Value,
51};
52pub use crate::ast::analyze::AnalyzeTarget;
53pub use crate::ast::ddl::{
54    AlterIndexOperation, AlterSinkOperation, AlterSourceOperation, AlterSubscriptionOperation,
55    AlterViewOperation,
56};
57use crate::keywords::Keyword;
58use crate::parser::{IncludeOption, IncludeOptionItem, Parser, ParserError, StrError};
59pub use crate::quote_ident::QuoteIdent;
60use crate::tokenizer::Tokenizer;
61
62/// Case-insensitive SQL option redaction with exact-name exemptions.
63#[derive(Debug, Default, serde::Serialize)]
64pub struct SqlOptionRedaction {
65    pub keywords: HashSet<String>,
66    pub allowlist: HashSet<String>,
67}
68
69impl SqlOptionRedaction {
70    pub fn new(
71        keywords: impl IntoIterator<Item = String>,
72        allowlist: impl IntoIterator<Item = String>,
73    ) -> Self {
74        Self {
75            keywords: keywords.into_iter().map(|s| s.to_lowercase()).collect(),
76            allowlist: allowlist.into_iter().map(|s| s.to_lowercase()).collect(),
77        }
78    }
79}
80
81pub type SqlOptionRedactionRef = Arc<SqlOptionRedaction>;
82
83task_local::task_local! {
84    pub static SQL_OPTION_REDACTION: SqlOptionRedactionRef;
85}
86
87pub struct DisplaySeparated<'a, T>
88where
89    T: fmt::Display,
90{
91    slice: &'a [T],
92    sep: &'static str,
93}
94
95impl<T> fmt::Display for DisplaySeparated<'_, T>
96where
97    T: fmt::Display,
98{
99    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
100        let mut delim = "";
101        for t in self.slice {
102            write!(f, "{}", delim)?;
103            delim = self.sep;
104            write!(f, "{}", t)?;
105        }
106        Ok(())
107    }
108}
109
110pub fn display_separated<'a, T>(slice: &'a [T], sep: &'static str) -> DisplaySeparated<'a, T>
111where
112    T: fmt::Display,
113{
114    DisplaySeparated { slice, sep }
115}
116
117pub fn display_comma_separated<T>(slice: &[T]) -> DisplaySeparated<'_, T>
118where
119    T: fmt::Display,
120{
121    DisplaySeparated { slice, sep: ", " }
122}
123
124/// An identifier, decomposed into its value or character data and the quote style.
125#[derive(Debug, Clone, PartialEq, Eq, Hash)]
126pub struct Ident {
127    /// The value of the identifier without quotes.
128    pub(crate) value: String,
129    /// The starting quote if any. Valid quote characters are the single quote,
130    /// double quote, backtick, and opening square bracket.
131    pub(crate) quote_style: Option<char>,
132}
133
134impl Ident {
135    /// Create a new identifier with the given value and no quotes.
136    /// the given value must not be a empty string.
137    // FIXME: should avoid using this function unless it's a literal or for testing.
138    pub fn new_unchecked<S>(value: S) -> Self
139    where
140        S: Into<String>,
141    {
142        Ident {
143            value: value.into(),
144            quote_style: None,
145        }
146    }
147
148    /// Create a new quoted identifier with the given quote and value.
149    /// the given value must not be a empty string and the given quote must be in ['\'', '"', '`',
150    /// '['].
151    pub fn with_quote_unchecked<S>(quote: char, value: S) -> Self
152    where
153        S: Into<String>,
154    {
155        Ident {
156            value: value.into(),
157            quote_style: Some(quote),
158        }
159    }
160
161    /// Create a new quoted identifier with the given quote and value.
162    /// returns ParserError when the given string is empty or the given quote is illegal.
163    pub fn with_quote_check<S>(quote: char, value: S) -> Result<Ident, ParserError>
164    where
165        S: Into<String>,
166    {
167        let value_str = value.into();
168        if value_str.is_empty() {
169            return Err(ParserError::ParserError(format!(
170                "zero-length delimited identifier at or near \"{value_str}\""
171            )));
172        }
173
174        if !(quote == '\'' || quote == '"' || quote == '`' || quote == '[') {
175            return Err(ParserError::ParserError(
176                "unexpected quote style".to_owned(),
177            ));
178        }
179
180        Ok(Ident {
181            value: value_str,
182            quote_style: Some(quote),
183        })
184    }
185
186    /// Returns the identifier value used for name lookup and comparison.
187    ///
188    /// Unquoted identifiers are folded to lowercase, while double-quoted identifiers preserve
189    /// their case and unescaped contents. For example, `Foo` becomes `foo`, while `"Foo"` remains
190    /// `Foo` and `"a""b"` becomes `a"b`.
191    pub fn real_value(&self) -> String {
192        match self.quote_style {
193            Some('"') => self.value.clone(),
194            _ => self.value.to_lowercase(),
195        }
196    }
197
198    /// Creates an identifier from a name used internally by the database.
199    ///
200    /// The input is stored as its unescaped value. Double quotes are selected when needed to
201    /// preserve the name in SQL; escaping embedded quotes is left to the [`Display`] implementation.
202    /// This behaves like the SQL `quote_ident` function and the [`QuoteIdent`] wrapper.
203    pub fn from_real_value(value: &str) -> Self {
204        let needs_quotes = QuoteIdent::needs_quotes(value);
205
206        if needs_quotes {
207            Self::with_quote_unchecked('"', value)
208        } else {
209            Self::new_unchecked(value)
210        }
211    }
212
213    pub fn quote_style(&self) -> Option<char> {
214        self.quote_style
215    }
216}
217
218impl From<&str> for Ident {
219    fn from(value: &str) -> Self {
220        Self::from_real_value(value)
221    }
222}
223
224impl ParseTo for Ident {
225    fn parse_to(parser: &mut Parser<'_>) -> ModalResult<Self> {
226        parser.parse_identifier()
227    }
228}
229
230impl fmt::Display for Ident {
231    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
232        match self.quote_style {
233            Some(q) if q == '\'' || q == '`' => write!(f, "{}{}{}", q, self.value, q),
234            Some('"') => write!(f, "\"{}\"", self.value.replace('"', "\"\"")),
235            Some('[') => write!(f, "[{}]", self.value),
236            None => f.write_str(&self.value),
237            _ => panic!("unexpected quote style"),
238        }
239    }
240}
241
242/// A name of a table, view, custom type, etc., possibly multi-part, i.e. db.schema.obj
243///
244/// Is is ensured to be non-empty.
245#[derive(Debug, Clone, PartialEq, Eq, Hash)]
246pub struct ObjectName(pub Vec<Ident>);
247
248impl ObjectName {
249    pub fn real_value(&self) -> String {
250        self.0
251            .iter()
252            .map(|ident| ident.real_value())
253            .collect::<Vec<_>>()
254            .join(".")
255    }
256
257    pub fn from_test_str(s: &str) -> Self {
258        ObjectName::from(vec![s.into()])
259    }
260
261    pub fn base_name(&self) -> String {
262        self.0
263            .iter()
264            .last()
265            .expect("should have base name")
266            .real_value()
267    }
268}
269
270impl fmt::Display for ObjectName {
271    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
272        write!(f, "{}", display_separated(&self.0, "."))
273    }
274}
275
276impl ParseTo for ObjectName {
277    fn parse_to(p: &mut Parser<'_>) -> ModalResult<Self> {
278        p.parse_object_name()
279    }
280}
281
282impl From<Vec<Ident>> for ObjectName {
283    fn from(value: Vec<Ident>) -> Self {
284        Self(value)
285    }
286}
287
288/// For array type `ARRAY[..]` or `[..]`
289#[derive(Debug, Clone, PartialEq, Eq, Hash)]
290pub struct Array {
291    /// The list of expressions between brackets
292    pub elem: Vec<Expr>,
293
294    /// `true` for  `ARRAY[..]`, `false` for `[..]`
295    pub named: bool,
296}
297
298impl fmt::Display for Array {
299    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
300        write!(
301            f,
302            "{}[{}]",
303            if self.named { "ARRAY" } else { "" },
304            display_comma_separated(&self.elem)
305        )
306    }
307}
308
309/// An escape character, to represent '' or a single character.
310#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
311pub struct EscapeChar(Option<char>);
312
313impl EscapeChar {
314    pub fn escape(ch: char) -> Self {
315        Self(Some(ch))
316    }
317
318    pub fn empty() -> Self {
319        Self(None)
320    }
321}
322
323impl fmt::Display for EscapeChar {
324    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
325        match self.0 {
326            Some(ch) => write!(f, "{}", ch),
327            None => f.write_str(""),
328        }
329    }
330}
331
332/// An SQL expression of any type.
333///
334/// The parser does not distinguish between expressions of different types
335/// (e.g. boolean vs string), so the caller must handle expressions of
336/// inappropriate type, like `WHERE 1` or `SELECT 1=1`, as necessary.
337#[derive(Debug, Clone, PartialEq, Eq, Hash)]
338pub enum Expr {
339    /// Identifier e.g. table name or column name
340    Identifier(Ident),
341    /// Multi-part identifier, e.g. `table_alias.column` or `schema.table.col`
342    CompoundIdentifier(Vec<Ident>),
343    /// Struct-field identifier.
344    /// Expr is an arbitrary expression, returning either a table or a column.
345    /// Idents are consecutive field accesses.
346    /// e.g. `(table.v1).v2` or `(table).v1.v2`
347    ///
348    /// It must contain parentheses to be distinguished from a [`Expr::CompoundIdentifier`].
349    /// See also <https://www.postgresql.org/docs/current/rowtypes.html#ROWTYPES-ACCESSING>
350    ///
351    /// The left parentheses must be put at the beginning of the expression.
352    /// The first parenthesized part is the `expr` part, and the rest are flattened into `idents`.
353    /// e.g., `((v1).v2.v3).v4` is equivalent to `(v1).v2.v3.v4`.
354    FieldIdentifier(Box<Expr>, Vec<Ident>),
355    /// `IS NULL` operator
356    IsNull(Box<Expr>),
357    /// `IS NOT NULL` operator
358    IsNotNull(Box<Expr>),
359    /// `IS TRUE` operator
360    IsTrue(Box<Expr>),
361    /// `IS NOT TRUE` operator
362    IsNotTrue(Box<Expr>),
363    /// `IS FALSE` operator
364    IsFalse(Box<Expr>),
365    /// `IS NOT FALSE` operator
366    IsNotFalse(Box<Expr>),
367    /// `IS UNKNOWN` operator
368    IsUnknown(Box<Expr>),
369    /// `IS NOT UNKNOWN` operator
370    IsNotUnknown(Box<Expr>),
371    /// `IS DISTINCT FROM` operator
372    IsDistinctFrom(Box<Expr>, Box<Expr>),
373    /// `IS NOT DISTINCT FROM` operator
374    IsNotDistinctFrom(Box<Expr>, Box<Expr>),
375    /// ```text
376    /// IS [ NOT ] JSON [ VALUE | ARRAY | OBJECT | SCALAR ]
377    /// [ { WITH | WITHOUT } UNIQUE [ KEYS ] ]
378    /// ```
379    IsJson {
380        expr: Box<Expr>,
381        negated: bool,
382        item_type: JsonPredicateType,
383        unique_keys: bool,
384    },
385    /// `[ NOT ] IN (val1, val2, ...)`
386    InList {
387        expr: Box<Expr>,
388        list: Vec<Expr>,
389        negated: bool,
390    },
391    /// `[ NOT ] IN (SELECT ...)`
392    InSubquery {
393        expr: Box<Expr>,
394        subquery: Box<Query>,
395        negated: bool,
396    },
397    /// `<expr> [ NOT ] BETWEEN <low> AND <high>`
398    Between {
399        expr: Box<Expr>,
400        negated: bool,
401        low: Box<Expr>,
402        high: Box<Expr>,
403    },
404    /// LIKE
405    Like {
406        negated: bool,
407        expr: Box<Expr>,
408        pattern: Box<Expr>,
409        escape_char: Option<EscapeChar>,
410    },
411    /// ILIKE (case-insensitive LIKE)
412    ILike {
413        negated: bool,
414        expr: Box<Expr>,
415        pattern: Box<Expr>,
416        escape_char: Option<EscapeChar>,
417    },
418    /// `<expr> [ NOT ] SIMILAR TO <pat> ESCAPE <esc_text>`
419    SimilarTo {
420        negated: bool,
421        expr: Box<Expr>,
422        pattern: Box<Expr>,
423        escape_char: Option<EscapeChar>,
424    },
425    /// Binary operation e.g. `1 + 1` or `foo > bar`
426    BinaryOp {
427        left: Box<Expr>,
428        op: BinaryOperator,
429        right: Box<Expr>,
430    },
431    /// Some operation e.g. `foo > Some(bar)`, It will be wrapped in the right side of BinaryExpr
432    SomeOp(Box<Expr>),
433    /// ALL operation e.g. `foo > ALL(bar)`, It will be wrapped in the right side of BinaryExpr
434    AllOp(Box<Expr>),
435    /// Unary operation e.g. `NOT foo`
436    UnaryOp {
437        op: UnaryOperator,
438        expr: Box<Expr>,
439    },
440    /// CAST an expression to a different data type e.g. `CAST(foo AS VARCHAR)`
441    Cast {
442        expr: Box<Expr>,
443        data_type: DataType,
444    },
445    /// TRY_CAST an expression to a different data type e.g. `TRY_CAST(foo AS VARCHAR)`
446    //  this differs from CAST in the choice of how to implement invalid conversions
447    TryCast {
448        expr: Box<Expr>,
449        data_type: DataType,
450    },
451    /// AT TIME ZONE converts `timestamp without time zone` to/from `timestamp with time zone` with
452    /// explicitly specified zone
453    AtTimeZone {
454        timestamp: Box<Expr>,
455        time_zone: Box<Expr>,
456    },
457    /// `EXTRACT(DateTimeField FROM <expr>)`
458    Extract {
459        field: String,
460        expr: Box<Expr>,
461    },
462    /// `SUBSTRING(<expr> [FROM <expr>] [FOR <expr>])`
463    Substring {
464        expr: Box<Expr>,
465        substring_from: Option<Box<Expr>>,
466        substring_for: Option<Box<Expr>>,
467    },
468    /// `POSITION(<expr> IN <expr>)`
469    Position {
470        substring: Box<Expr>,
471        string: Box<Expr>,
472    },
473    /// `OVERLAY(<expr> PLACING <expr> FROM <expr> [ FOR <expr> ])`
474    Overlay {
475        expr: Box<Expr>,
476        new_substring: Box<Expr>,
477        start: Box<Expr>,
478        count: Option<Box<Expr>>,
479    },
480    /// `TRIM([BOTH | LEADING | TRAILING] [<expr>] FROM <expr>)`\
481    /// Or\
482    /// `TRIM([BOTH | LEADING | TRAILING] [FROM] <expr> [, <expr>])`
483    Trim {
484        expr: Box<Expr>,
485        // ([BOTH | LEADING | TRAILING], <expr>)
486        trim_where: Option<TrimWhereField>,
487        trim_what: Option<Box<Expr>>,
488    },
489    /// `expr COLLATE collation`
490    Collate {
491        expr: Box<Expr>,
492        collation: ObjectName,
493    },
494    /// Nested expression e.g. `(foo > bar)` or `(1)`
495    Nested(Box<Expr>),
496    /// A literal value, such as string, number, date or NULL
497    Value(Value),
498    /// Parameter Symbol e.g. `$1`, `$1::int`
499    Parameter {
500        index: u64,
501    },
502    /// A constant of form `<data_type> 'value'`.
503    /// This can represent ANSI SQL `DATE`, `TIME`, and `TIMESTAMP` literals (such as `DATE
504    /// '2020-01-01'`), as well as constants of other types (a non-standard PostgreSQL extension).
505    TypedString {
506        data_type: DataType,
507        value: String,
508    },
509    /// Scalar function call e.g. `LEFT(foo, 5)`
510    Function(Function),
511    /// `CASE [<operand>] WHEN <condition> THEN <result> ... [ELSE <result>] END`
512    ///
513    /// Note we only recognize a complete single expression as `<condition>`,
514    /// not `< 0` nor `1, 2, 3` as allowed in a `<simple when clause>` per
515    /// <https://jakewheat.github.io/sql-overview/sql-2011-foundation-grammar.html#simple-when-clause>
516    Case {
517        operand: Option<Box<Expr>>,
518        conditions: Vec<Expr>,
519        results: Vec<Expr>,
520        else_result: Option<Box<Expr>>,
521    },
522    /// An exists expression `EXISTS(SELECT ...)`, used in expressions like
523    /// `WHERE EXISTS (SELECT ...)`.
524    Exists(Box<Query>),
525    /// A parenthesized subquery `(SELECT ...)`, used in expression like
526    /// `SELECT (subquery) AS x` or `WHERE (subquery) = x`
527    Subquery(Box<Query>),
528    /// The `GROUPING SETS` expr.
529    GroupingSets(Vec<Vec<Expr>>),
530    /// The `CUBE` expr.
531    Cube(Vec<Vec<Expr>>),
532    /// The `ROLLUP` expr.
533    Rollup(Vec<Vec<Expr>>),
534    /// The `ROW` expr. The `ROW` keyword can be omitted,
535    Row(Vec<Expr>),
536    /// An array constructor `ARRAY[[2,3,4],[5,6,7]]`
537    Array(Array),
538    /// An array constructing subquery `ARRAY(SELECT 2 UNION SELECT 3)`
539    ArraySubquery(Box<Query>),
540    /// A subscript expression `arr[1]` or `map['a']`
541    Index {
542        obj: Box<Expr>,
543        index: Box<Expr>,
544    },
545    /// A slice expression `arr[1:3]`
546    ArrayRangeIndex {
547        obj: Box<Expr>,
548        start: Option<Box<Expr>>,
549        end: Option<Box<Expr>>,
550    },
551    LambdaFunction {
552        args: Vec<Ident>,
553        body: Box<Expr>,
554    },
555    Map {
556        entries: Vec<(Expr, Expr)>,
557    },
558}
559
560impl fmt::Display for Expr {
561    #[expect(clippy::disallowed_methods, reason = "use zip_eq")]
562    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
563        match self {
564            Expr::Identifier(s) => write!(f, "{}", s),
565            Expr::CompoundIdentifier(s) => write!(f, "{}", display_separated(s, ".")),
566            Expr::FieldIdentifier(ast, s) => write!(f, "({}).{}", ast, display_separated(s, ".")),
567            Expr::IsNull(ast) => write!(f, "{} IS NULL", ast),
568            Expr::IsNotNull(ast) => write!(f, "{} IS NOT NULL", ast),
569            Expr::IsTrue(ast) => write!(f, "{} IS TRUE", ast),
570            Expr::IsNotTrue(ast) => write!(f, "{} IS NOT TRUE", ast),
571            Expr::IsFalse(ast) => write!(f, "{} IS FALSE", ast),
572            Expr::IsNotFalse(ast) => write!(f, "{} IS NOT FALSE", ast),
573            Expr::IsUnknown(ast) => write!(f, "{} IS UNKNOWN", ast),
574            Expr::IsNotUnknown(ast) => write!(f, "{} IS NOT UNKNOWN", ast),
575            Expr::IsJson {
576                expr,
577                negated,
578                item_type,
579                unique_keys,
580            } => write!(
581                f,
582                "{} IS {}JSON{}{}",
583                expr,
584                if *negated { "NOT " } else { "" },
585                item_type,
586                if *unique_keys {
587                    " WITH UNIQUE KEYS"
588                } else {
589                    ""
590                },
591            ),
592            Expr::InList {
593                expr,
594                list,
595                negated,
596            } => write!(
597                f,
598                "{} {}IN ({})",
599                expr,
600                if *negated { "NOT " } else { "" },
601                display_comma_separated(list)
602            ),
603            Expr::InSubquery {
604                expr,
605                subquery,
606                negated,
607            } => write!(
608                f,
609                "{} {}IN ({})",
610                expr,
611                if *negated { "NOT " } else { "" },
612                subquery
613            ),
614            Expr::Between {
615                expr,
616                negated,
617                low,
618                high,
619            } => write!(
620                f,
621                "{} {}BETWEEN {} AND {}",
622                expr,
623                if *negated { "NOT " } else { "" },
624                low,
625                high
626            ),
627            Expr::Like {
628                negated,
629                expr,
630                pattern,
631                escape_char,
632            } => match escape_char {
633                Some(ch) => write!(
634                    f,
635                    "{} {}LIKE {} ESCAPE '{}'",
636                    expr,
637                    if *negated { "NOT " } else { "" },
638                    pattern,
639                    ch
640                ),
641                _ => write!(
642                    f,
643                    "{} {}LIKE {}",
644                    expr,
645                    if *negated { "NOT " } else { "" },
646                    pattern
647                ),
648            },
649            Expr::ILike {
650                negated,
651                expr,
652                pattern,
653                escape_char,
654            } => match escape_char {
655                Some(ch) => write!(
656                    f,
657                    "{} {}ILIKE {} ESCAPE '{}'",
658                    expr,
659                    if *negated { "NOT " } else { "" },
660                    pattern,
661                    ch
662                ),
663                _ => write!(
664                    f,
665                    "{} {}ILIKE {}",
666                    expr,
667                    if *negated { "NOT " } else { "" },
668                    pattern
669                ),
670            },
671            Expr::SimilarTo {
672                negated,
673                expr,
674                pattern,
675                escape_char,
676            } => match escape_char {
677                Some(ch) => write!(
678                    f,
679                    "{} {}SIMILAR TO {} ESCAPE '{}'",
680                    expr,
681                    if *negated { "NOT " } else { "" },
682                    pattern,
683                    ch
684                ),
685                _ => write!(
686                    f,
687                    "{} {}SIMILAR TO {}",
688                    expr,
689                    if *negated { "NOT " } else { "" },
690                    pattern
691                ),
692            },
693            Expr::BinaryOp { left, op, right } => write!(f, "{} {} {}", left, op, right),
694            Expr::SomeOp(expr) => write!(f, "SOME({})", expr),
695            Expr::AllOp(expr) => write!(f, "ALL({})", expr),
696            Expr::UnaryOp { op, expr } => {
697                write!(f, "{} {}", op, expr)
698            }
699            Expr::Cast { expr, data_type } => write!(f, "CAST({} AS {})", expr, data_type),
700            Expr::TryCast { expr, data_type } => write!(f, "TRY_CAST({} AS {})", expr, data_type),
701            Expr::AtTimeZone {
702                timestamp,
703                time_zone,
704            } => write!(f, "{} AT TIME ZONE {}", timestamp, time_zone),
705            Expr::Extract { field, expr } => write!(f, "EXTRACT({} FROM {})", field, expr),
706            Expr::Collate { expr, collation } => write!(f, "{} COLLATE {}", expr, collation),
707            Expr::Nested(ast) => write!(f, "({})", ast),
708            Expr::Value(v) => write!(f, "{}", v),
709            Expr::Parameter { index } => write!(f, "${}", index),
710            Expr::TypedString { data_type, value } => {
711                write!(f, "{}", data_type)?;
712                write!(f, " '{}'", value::escape_single_quote_string(value))
713            }
714            Expr::Function(fun) => write!(f, "{}", fun),
715            Expr::Case {
716                operand,
717                conditions,
718                results,
719                else_result,
720            } => {
721                write!(f, "CASE")?;
722                if let Some(operand) = operand {
723                    write!(f, " {}", operand)?;
724                }
725                for (c, r) in conditions.iter().zip_eq(results) {
726                    write!(f, " WHEN {} THEN {}", c, r)?;
727                }
728
729                if let Some(else_result) = else_result {
730                    write!(f, " ELSE {}", else_result)?;
731                }
732                write!(f, " END")
733            }
734            Expr::Exists(s) => write!(f, "EXISTS ({})", s),
735            Expr::Subquery(s) => write!(f, "({})", s),
736            Expr::GroupingSets(sets) => {
737                write!(f, "GROUPING SETS (")?;
738                let mut sep = "";
739                for set in sets {
740                    write!(f, "{}", sep)?;
741                    sep = ", ";
742                    write!(f, "({})", display_comma_separated(set))?;
743                }
744                write!(f, ")")
745            }
746            Expr::Cube(sets) => {
747                write!(f, "CUBE (")?;
748                let mut sep = "";
749                for set in sets {
750                    write!(f, "{}", sep)?;
751                    sep = ", ";
752                    if set.len() == 1 {
753                        write!(f, "{}", set[0])?;
754                    } else {
755                        write!(f, "({})", display_comma_separated(set))?;
756                    }
757                }
758                write!(f, ")")
759            }
760            Expr::Rollup(sets) => {
761                write!(f, "ROLLUP (")?;
762                let mut sep = "";
763                for set in sets {
764                    write!(f, "{}", sep)?;
765                    sep = ", ";
766                    if set.len() == 1 {
767                        write!(f, "{}", set[0])?;
768                    } else {
769                        write!(f, "({})", display_comma_separated(set))?;
770                    }
771                }
772                write!(f, ")")
773            }
774            Expr::Substring {
775                expr,
776                substring_from,
777                substring_for,
778            } => {
779                write!(f, "SUBSTRING({}", expr)?;
780                if let Some(from_part) = substring_from {
781                    write!(f, " FROM {}", from_part)?;
782                }
783                if let Some(from_part) = substring_for {
784                    write!(f, " FOR {}", from_part)?;
785                }
786
787                write!(f, ")")
788            }
789            Expr::Position { substring, string } => {
790                write!(f, "POSITION({} IN {})", substring, string)
791            }
792            Expr::Overlay {
793                expr,
794                new_substring,
795                start,
796                count,
797            } => {
798                write!(f, "OVERLAY({}", expr)?;
799                write!(f, " PLACING {}", new_substring)?;
800                write!(f, " FROM {}", start)?;
801
802                if let Some(count_expr) = count {
803                    write!(f, " FOR {}", count_expr)?;
804                }
805
806                write!(f, ")")
807            }
808            Expr::IsDistinctFrom(a, b) => write!(f, "{} IS DISTINCT FROM {}", a, b),
809            Expr::IsNotDistinctFrom(a, b) => write!(f, "{} IS NOT DISTINCT FROM {}", a, b),
810            Expr::Trim {
811                expr,
812                trim_where,
813                trim_what,
814            } => {
815                write!(f, "TRIM(")?;
816                if let Some(ident) = trim_where {
817                    write!(f, "{} ", ident)?;
818                }
819                if let Some(trim_char) = trim_what {
820                    write!(f, "{} ", trim_char)?;
821                }
822                write!(f, "FROM {})", expr)
823            }
824            Expr::Row(exprs) => write!(
825                f,
826                "ROW({})",
827                exprs
828                    .iter()
829                    .map(|v| v.to_string())
830                    .collect::<Vec<String>>()
831                    .as_slice()
832                    .join(", ")
833            ),
834            Expr::Index { obj, index } => {
835                write!(f, "{}[{}]", obj, index)?;
836                Ok(())
837            }
838            Expr::ArrayRangeIndex { obj, start, end } => {
839                let start_str = match start {
840                    None => "".to_owned(),
841                    Some(start) => format!("{}", start),
842                };
843                let end_str = match end {
844                    None => "".to_owned(),
845                    Some(end) => format!("{}", end),
846                };
847                write!(f, "{}[{}:{}]", obj, start_str, end_str)?;
848                Ok(())
849            }
850            Expr::Array(exprs) => write!(f, "{}", exprs),
851            Expr::ArraySubquery(s) => write!(f, "ARRAY ({})", s),
852            Expr::LambdaFunction { args, body } => {
853                write!(
854                    f,
855                    "|{}| {}",
856                    args.iter().map(ToString::to_string).join(", "),
857                    body
858                )
859            }
860            Expr::Map { entries } => {
861                write!(
862                    f,
863                    "MAP {{{}}}",
864                    entries
865                        .iter()
866                        .map(|(k, v)| format!("{}: {}", k, v))
867                        .join(", ")
868                )
869            }
870        }
871    }
872}
873
874/// A window specification (i.e. `OVER (PARTITION BY .. ORDER BY .. etc.)`).
875/// This is used both for named window definitions and inline window specifications.
876#[derive(Debug, Clone, PartialEq, Eq, Hash)]
877pub struct WindowSpec {
878    pub partition_by: Vec<Expr>,
879    pub order_by: Vec<OrderByExpr>,
880    pub window_frame: Option<WindowFrame>,
881}
882
883/// A window definition that can appear in the OVER clause of a window function.
884/// This can be either an inline window specification or a reference to a named window.
885#[derive(Debug, Clone, PartialEq, Eq, Hash)]
886pub enum Window {
887    /// Inline window specification: `OVER (PARTITION BY ... ORDER BY ...)`
888    Spec(WindowSpec),
889    /// Named window reference: `OVER window_name`
890    Name(Ident),
891}
892
893impl fmt::Display for WindowSpec {
894    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
895        let mut delim = "";
896        if !self.partition_by.is_empty() {
897            delim = " ";
898            write!(
899                f,
900                "PARTITION BY {}",
901                display_comma_separated(&self.partition_by)
902            )?;
903        }
904        if !self.order_by.is_empty() {
905            f.write_str(delim)?;
906            delim = " ";
907            write!(f, "ORDER BY {}", display_comma_separated(&self.order_by))?;
908        }
909        if let Some(window_frame) = &self.window_frame {
910            f.write_str(delim)?;
911            window_frame.fmt(f)?;
912        }
913        Ok(())
914    }
915}
916
917impl fmt::Display for Window {
918    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
919        match self {
920            Window::Spec(spec) => write!(f, "({})", spec),
921            Window::Name(name) => write!(f, "{}", name),
922        }
923    }
924}
925
926/// Specifies the data processed by a window function, e.g.
927/// `RANGE UNBOUNDED PRECEDING` or `ROWS BETWEEN 5 PRECEDING AND CURRENT ROW`.
928///
929/// Note: The parser does not validate the specified bounds; the caller should
930/// reject invalid bounds like `ROWS UNBOUNDED FOLLOWING` before execution.
931#[derive(Debug, Clone, PartialEq, Eq, Hash)]
932pub struct WindowFrame {
933    pub units: WindowFrameUnits,
934    pub bounds: WindowFrameBounds,
935    pub exclusion: Option<WindowFrameExclusion>,
936}
937
938#[derive(Debug, Clone, PartialEq, Eq, Hash)]
939pub enum WindowFrameUnits {
940    Rows,
941    Range,
942    Groups,
943    Session,
944}
945
946#[derive(Debug, Clone, PartialEq, Eq, Hash)]
947pub enum WindowFrameBounds {
948    Bounds {
949        start: WindowFrameBound,
950        /// The right bound of the `BETWEEN .. AND` clause. The end bound of `None`
951        /// indicates the shorthand form (e.g. `ROWS 1 PRECEDING`), which must
952        /// behave the same as `end_bound = WindowFrameBound::CurrentRow`.
953        end: Option<WindowFrameBound>,
954    },
955    Gap(Box<Expr>),
956}
957
958impl fmt::Display for WindowFrame {
959    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
960        write!(f, "{} ", self.units)?;
961        match &self.bounds {
962            WindowFrameBounds::Bounds { start, end } => {
963                if let Some(end) = end {
964                    write!(f, "BETWEEN {} AND {}", start, end)
965                } else {
966                    write!(f, "{}", start)
967                }
968            }
969            WindowFrameBounds::Gap(gap) => {
970                write!(f, "WITH GAP {}", gap)
971            }
972        }
973    }
974}
975
976impl fmt::Display for WindowFrameUnits {
977    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
978        f.write_str(match self {
979            WindowFrameUnits::Rows => "ROWS",
980            WindowFrameUnits::Range => "RANGE",
981            WindowFrameUnits::Groups => "GROUPS",
982            WindowFrameUnits::Session => "SESSION",
983        })
984    }
985}
986
987/// Specifies [WindowFrame]'s `start_bound` and `end_bound`
988#[derive(Debug, Clone, PartialEq, Eq, Hash)]
989pub enum WindowFrameBound {
990    /// `CURRENT ROW`
991    CurrentRow,
992    /// `<offset> PRECEDING` or `UNBOUNDED PRECEDING`
993    Preceding(Option<Box<Expr>>),
994    /// `<offset> FOLLOWING` or `UNBOUNDED FOLLOWING`.
995    Following(Option<Box<Expr>>),
996}
997
998impl fmt::Display for WindowFrameBound {
999    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1000        match self {
1001            WindowFrameBound::CurrentRow => f.write_str("CURRENT ROW"),
1002            WindowFrameBound::Preceding(None) => f.write_str("UNBOUNDED PRECEDING"),
1003            WindowFrameBound::Following(None) => f.write_str("UNBOUNDED FOLLOWING"),
1004            WindowFrameBound::Preceding(Some(n)) => write!(f, "{} PRECEDING", n),
1005            WindowFrameBound::Following(Some(n)) => write!(f, "{} FOLLOWING", n),
1006        }
1007    }
1008}
1009
1010/// Frame exclusion option of [WindowFrame].
1011#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
1012pub enum WindowFrameExclusion {
1013    CurrentRow,
1014    Group,
1015    Ties,
1016    NoOthers,
1017}
1018
1019impl fmt::Display for WindowFrameExclusion {
1020    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1021        match self {
1022            WindowFrameExclusion::CurrentRow => f.write_str("EXCLUDE CURRENT ROW"),
1023            WindowFrameExclusion::Group => f.write_str("EXCLUDE GROUP"),
1024            WindowFrameExclusion::Ties => f.write_str("EXCLUDE TIES"),
1025            WindowFrameExclusion::NoOthers => f.write_str("EXCLUDE NO OTHERS"),
1026        }
1027    }
1028}
1029
1030#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1031pub enum AddDropSync {
1032    ADD,
1033    DROP,
1034    SYNC,
1035}
1036
1037impl fmt::Display for AddDropSync {
1038    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1039        match self {
1040            AddDropSync::SYNC => f.write_str("SYNC PARTITIONS"),
1041            AddDropSync::DROP => f.write_str("DROP PARTITIONS"),
1042            AddDropSync::ADD => f.write_str("ADD PARTITIONS"),
1043        }
1044    }
1045}
1046
1047#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1048pub enum ShowObject {
1049    Table { schema: Option<Ident> },
1050    InternalTable { schema: Option<Ident> },
1051    Database,
1052    Schema,
1053    View { schema: Option<Ident> },
1054    MaterializedView { schema: Option<Ident> },
1055    Source { schema: Option<Ident> },
1056    Sink { schema: Option<Ident> },
1057    Subscription { schema: Option<Ident> },
1058    Columns { table: ObjectName },
1059    Connection { schema: Option<Ident> },
1060    Secret { schema: Option<Ident> },
1061    Function { schema: Option<Ident> },
1062    Indexes { table: ObjectName },
1063    Cluster,
1064    Jobs,
1065    ProcessList,
1066    Cursor,
1067    SubscriptionCursor,
1068}
1069
1070#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1071pub struct JobIdents(pub Vec<u32>);
1072
1073impl fmt::Display for ShowObject {
1074    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1075        fn fmt_schema(schema: &Option<Ident>) -> String {
1076            if let Some(schema) = schema {
1077                format!(" FROM {}", schema.value)
1078            } else {
1079                "".to_owned()
1080            }
1081        }
1082
1083        match self {
1084            ShowObject::Database => f.write_str("DATABASES"),
1085            ShowObject::Schema => f.write_str("SCHEMAS"),
1086            ShowObject::Table { schema } => {
1087                write!(f, "TABLES{}", fmt_schema(schema))
1088            }
1089            ShowObject::InternalTable { schema } => {
1090                write!(f, "INTERNAL TABLES{}", fmt_schema(schema))
1091            }
1092            ShowObject::View { schema } => {
1093                write!(f, "VIEWS{}", fmt_schema(schema))
1094            }
1095            ShowObject::MaterializedView { schema } => {
1096                write!(f, "MATERIALIZED VIEWS{}", fmt_schema(schema))
1097            }
1098            ShowObject::Source { schema } => write!(f, "SOURCES{}", fmt_schema(schema)),
1099            ShowObject::Sink { schema } => write!(f, "SINKS{}", fmt_schema(schema)),
1100            ShowObject::Columns { table } => write!(f, "COLUMNS FROM {}", table),
1101            ShowObject::Connection { schema } => write!(f, "CONNECTIONS{}", fmt_schema(schema)),
1102            ShowObject::Function { schema } => write!(f, "FUNCTIONS{}", fmt_schema(schema)),
1103            ShowObject::Indexes { table } => write!(f, "INDEXES FROM {}", table),
1104            ShowObject::Cluster => {
1105                write!(f, "CLUSTER")
1106            }
1107            ShowObject::Jobs => write!(f, "JOBS"),
1108            ShowObject::ProcessList => write!(f, "PROCESSLIST"),
1109            ShowObject::Subscription { schema } => write!(f, "SUBSCRIPTIONS{}", fmt_schema(schema)),
1110            ShowObject::Secret { schema } => write!(f, "SECRETS{}", fmt_schema(schema)),
1111            ShowObject::Cursor => write!(f, "CURSORS"),
1112            ShowObject::SubscriptionCursor => write!(f, "SUBSCRIPTION CURSORS"),
1113        }
1114    }
1115}
1116
1117#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1118pub enum ShowCreateType {
1119    Table,
1120    MaterializedView,
1121    View,
1122    Index,
1123    Source,
1124    Sink,
1125    Function,
1126    Subscription,
1127}
1128
1129impl fmt::Display for ShowCreateType {
1130    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1131        match self {
1132            ShowCreateType::Table => f.write_str("TABLE"),
1133            ShowCreateType::MaterializedView => f.write_str("MATERIALIZED VIEW"),
1134            ShowCreateType::View => f.write_str("VIEW"),
1135            ShowCreateType::Index => f.write_str("INDEX"),
1136            ShowCreateType::Source => f.write_str("SOURCE"),
1137            ShowCreateType::Sink => f.write_str("SINK"),
1138            ShowCreateType::Function => f.write_str("FUNCTION"),
1139            ShowCreateType::Subscription => f.write_str("SUBSCRIPTION"),
1140        }
1141    }
1142}
1143
1144#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1145pub enum CommentObject {
1146    Column,
1147    Table,
1148}
1149
1150impl fmt::Display for CommentObject {
1151    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1152        match self {
1153            CommentObject::Column => f.write_str("COLUMN"),
1154            CommentObject::Table => f.write_str("TABLE"),
1155        }
1156    }
1157}
1158
1159#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
1160pub enum ExplainType {
1161    Logical,
1162    Physical,
1163    DistSql,
1164}
1165
1166impl fmt::Display for ExplainType {
1167    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1168        match self {
1169            ExplainType::Logical => f.write_str("Logical"),
1170            ExplainType::Physical => f.write_str("Physical"),
1171            ExplainType::DistSql => f.write_str("DistSQL"),
1172        }
1173    }
1174}
1175
1176#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
1177pub enum ExplainFormat {
1178    Text,
1179    Json,
1180    Xml,
1181    Yaml,
1182    Dot,
1183}
1184
1185impl fmt::Display for ExplainFormat {
1186    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1187        match self {
1188            ExplainFormat::Text => f.write_str("TEXT"),
1189            ExplainFormat::Json => f.write_str("JSON"),
1190            ExplainFormat::Xml => f.write_str("XML"),
1191            ExplainFormat::Yaml => f.write_str("YAML"),
1192            ExplainFormat::Dot => f.write_str("DOT"),
1193        }
1194    }
1195}
1196
1197#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
1198pub struct ExplainOptions {
1199    /// Display additional information regarding the plan.
1200    pub verbose: bool,
1201    // Trace plan transformation of the optimizer step by step
1202    pub trace: bool,
1203    // Display backfill order
1204    pub backfill: bool,
1205    // explain's plan type
1206    pub explain_type: ExplainType,
1207    // explain's plan format
1208    pub explain_format: ExplainFormat,
1209}
1210
1211impl Default for ExplainOptions {
1212    fn default() -> Self {
1213        Self {
1214            verbose: false,
1215            trace: false,
1216            backfill: false,
1217            explain_type: ExplainType::Physical,
1218            explain_format: ExplainFormat::Text,
1219        }
1220    }
1221}
1222
1223impl fmt::Display for ExplainOptions {
1224    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1225        let default = Self::default();
1226        if *self == default {
1227            Ok(())
1228        } else {
1229            let mut option_strs = vec![];
1230            if self.verbose {
1231                option_strs.push("VERBOSE".to_owned());
1232            }
1233            if self.trace {
1234                option_strs.push("TRACE".to_owned());
1235            }
1236            if self.backfill {
1237                option_strs.push("BACKFILL".to_owned());
1238            }
1239            if self.explain_type == default.explain_type {
1240                option_strs.push(self.explain_type.to_string());
1241            }
1242            if self.explain_format == default.explain_format {
1243                option_strs.push(self.explain_format.to_string());
1244            }
1245            write!(f, "{}", option_strs.iter().format(","))
1246        }
1247    }
1248}
1249
1250#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1251pub struct CdcTableInfo {
1252    pub source_name: ObjectName,
1253    pub external_table_name: String,
1254}
1255
1256#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1257pub enum CopyEntity {
1258    Query(Box<Query>),
1259    Table {
1260        /// TABLE
1261        table_name: ObjectName,
1262        /// COLUMNS
1263        columns: Vec<Ident>,
1264    },
1265}
1266
1267#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1268pub enum CopyTarget {
1269    Stdin {
1270        /// VALUES a vector of values to be copied
1271        values: Vec<Option<String>>,
1272    },
1273    Stdout,
1274}
1275
1276#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1277pub enum WaitTarget {
1278    All,
1279    Table(ObjectName),
1280    MaterializedView(ObjectName),
1281    Sink(ObjectName),
1282    Index(ObjectName),
1283}
1284
1285#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
1286pub enum FileCacheType {
1287    Meta,
1288    Data,
1289    All,
1290}
1291
1292/// A top-level statement (SELECT, INSERT, CREATE, etc.)
1293
1294#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1295pub enum Statement {
1296    /// Analyze (Hive)
1297    Analyze {
1298        table_name: ObjectName,
1299    },
1300    /// Truncate (Hive)
1301    Truncate {
1302        table_name: ObjectName,
1303    },
1304    /// Refresh table
1305    Refresh {
1306        table_name: ObjectName,
1307    },
1308    /// SELECT
1309    Query(Box<Query>),
1310    /// INSERT
1311    Insert {
1312        /// TABLE
1313        table_name: ObjectName,
1314        /// COLUMNS
1315        columns: Vec<Ident>,
1316        /// A SQL query that specifies what to insert
1317        source: Box<Query>,
1318        /// Define output of this insert statement
1319        returning: Vec<SelectItem>,
1320    },
1321    Copy {
1322        entity: CopyEntity,
1323        target: CopyTarget,
1324    },
1325    /// UPDATE
1326    Update {
1327        /// TABLE
1328        table_name: ObjectName,
1329        /// Column assignments
1330        assignments: Vec<Assignment>,
1331        /// WHERE
1332        selection: Option<Expr>,
1333        /// RETURNING
1334        returning: Vec<SelectItem>,
1335    },
1336    /// DELETE
1337    Delete {
1338        /// FROM
1339        table_name: ObjectName,
1340        /// WHERE
1341        selection: Option<Expr>,
1342        /// RETURNING
1343        returning: Vec<SelectItem>,
1344    },
1345    /// DELETE META SNAPSHOT(S)
1346    DeleteMetaSnapshots {
1347        snapshot_ids: Vec<u64>,
1348    },
1349    /// DISCARD
1350    Discard(DiscardType),
1351    /// CREATE VIEW
1352    CreateView {
1353        or_replace: bool,
1354        materialized: bool,
1355        if_not_exists: bool,
1356        /// View name
1357        name: ObjectName,
1358        columns: Vec<Ident>,
1359        query: Box<Query>,
1360        emit_mode: Option<EmitMode>,
1361        with_options: Vec<SqlOption>,
1362    },
1363    /// CREATE TABLE
1364    CreateTable {
1365        or_replace: bool,
1366        temporary: bool,
1367        if_not_exists: bool,
1368        /// Table name
1369        name: ObjectName,
1370        /// Optional schema
1371        columns: Vec<ColumnDef>,
1372        // The wildchar position in columns defined in sql. Only exist when using external schema.
1373        wildcard_idx: Option<usize>,
1374        constraints: Vec<TableConstraint>,
1375        with_options: Vec<SqlOption>,
1376        /// `FORMAT ... ENCODE ...` for table with connector
1377        format_encode: Option<CompatibleFormatEncode>,
1378        /// The watermark defined on source.
1379        source_watermarks: Vec<SourceWatermark>,
1380        /// Append only table.
1381        append_only: bool,
1382        /// On conflict behavior
1383        on_conflict: Option<OnConflict>,
1384        /// with_version_columns behind on conflict - supports multiple version columns
1385        with_version_columns: Vec<Ident>,
1386        /// `AS ( query )`
1387        query: Option<Box<Query>>,
1388        /// `FROM cdc_source TABLE database_name.table_name`
1389        cdc_table_info: Option<CdcTableInfo>,
1390        /// `INCLUDE a AS b INCLUDE c`
1391        include_column_options: IncludeOption,
1392        /// `VALIDATE SECRET secure_secret_name AS secure_compare ()`
1393        webhook_info: Option<WebhookSourceInfo>,
1394        /// `Engine = [hummock | iceberg]`
1395        engine: Engine,
1396    },
1397    /// CREATE INDEX
1398    CreateIndex {
1399        /// index name
1400        name: ObjectName,
1401        table_name: ObjectName,
1402        columns: Vec<OrderByExpr>,
1403        method: Option<Ident>,
1404        include: Vec<Ident>,
1405        distributed_by: Vec<Expr>,
1406        unique: bool,
1407        if_not_exists: bool,
1408        with_properties: WithProperties,
1409    },
1410    /// CREATE SOURCE
1411    CreateSource {
1412        stmt: CreateSourceStatement,
1413    },
1414    /// CREATE SINK
1415    CreateSink {
1416        stmt: CreateSinkStatement,
1417    },
1418    /// CREATE SUBSCRIPTION
1419    CreateSubscription {
1420        stmt: CreateSubscriptionStatement,
1421    },
1422    /// CREATE CONNECTION
1423    CreateConnection {
1424        stmt: CreateConnectionStatement,
1425    },
1426    CreateSecret {
1427        stmt: CreateSecretStatement,
1428    },
1429    /// CREATE FUNCTION
1430    ///
1431    /// Postgres: <https://www.postgresql.org/docs/15/sql-createfunction.html>
1432    CreateFunction {
1433        or_replace: bool,
1434        temporary: bool,
1435        if_not_exists: bool,
1436        name: ObjectName,
1437        args: Option<Vec<OperateFunctionArg>>,
1438        returns: Option<CreateFunctionReturns>,
1439        /// Optional parameters.
1440        params: CreateFunctionBody,
1441        with_options: CreateFunctionWithOptions, // FIXME(eric): use Option<>
1442    },
1443    /// CREATE AGGREGATE
1444    ///
1445    /// Postgres: <https://www.postgresql.org/docs/15/sql-createaggregate.html>
1446    CreateAggregate {
1447        or_replace: bool,
1448        if_not_exists: bool,
1449        name: ObjectName,
1450        args: Vec<OperateFunctionArg>,
1451        returns: DataType,
1452        /// Optional parameters.
1453        append_only: bool,
1454        params: CreateFunctionBody,
1455    },
1456
1457    /// DECLARE CURSOR
1458    DeclareCursor {
1459        stmt: DeclareCursorStatement,
1460    },
1461
1462    // FETCH CURSOR
1463    FetchCursor {
1464        stmt: FetchCursorStatement,
1465    },
1466
1467    // CLOSE CURSOR
1468    CloseCursor {
1469        stmt: CloseCursorStatement,
1470    },
1471
1472    /// ALTER DATABASE
1473    AlterDatabase {
1474        name: ObjectName,
1475        operation: AlterDatabaseOperation,
1476    },
1477    /// ALTER SCHEMA
1478    AlterSchema {
1479        name: ObjectName,
1480        operation: AlterSchemaOperation,
1481    },
1482    /// ALTER TABLE
1483    AlterTable {
1484        /// Table name
1485        name: ObjectName,
1486        operation: AlterTableOperation,
1487    },
1488    /// ALTER INDEX
1489    AlterIndex {
1490        /// Index name
1491        name: ObjectName,
1492        operation: AlterIndexOperation,
1493    },
1494    /// ALTER VIEW
1495    AlterView {
1496        /// View name
1497        name: ObjectName,
1498        materialized: bool,
1499        operation: AlterViewOperation,
1500    },
1501    /// ALTER SINK
1502    AlterSink {
1503        /// Sink name
1504        name: ObjectName,
1505        operation: AlterSinkOperation,
1506    },
1507    AlterSubscription {
1508        name: ObjectName,
1509        operation: AlterSubscriptionOperation,
1510    },
1511    /// ALTER SOURCE
1512    AlterSource {
1513        /// Source name
1514        name: ObjectName,
1515        operation: AlterSourceOperation,
1516    },
1517    /// ALTER FUNCTION
1518    AlterFunction {
1519        /// Function name
1520        name: ObjectName,
1521        args: Option<Vec<OperateFunctionArg>>,
1522        operation: AlterFunctionOperation,
1523    },
1524    /// ALTER CONNECTION
1525    AlterConnection {
1526        /// Connection name
1527        name: ObjectName,
1528        operation: AlterConnectionOperation,
1529    },
1530    /// ALTER SECRET
1531    AlterSecret {
1532        /// Secret name
1533        name: ObjectName,
1534        operation: AlterSecretOperation,
1535    },
1536    /// ALTER FRAGMENT
1537    AlterFragment {
1538        fragment_ids: Vec<u32>,
1539        operation: AlterFragmentOperation,
1540    },
1541    /// ALTER COMPACTION GROUP
1542    AlterCompactionGroup {
1543        group_ids: Vec<u64>,
1544        operation: AlterCompactionGroupOperation,
1545    },
1546    /// DESCRIBE relation
1547    /// ALTER DEFAULT PRIVILEGES
1548    AlterDefaultPrivileges {
1549        target_users: Option<Vec<Ident>>,
1550        schema_names: Option<Vec<ObjectName>>,
1551        operation: DefaultPrivilegeOperation,
1552    },
1553    /// DESCRIBE relation
1554    Describe {
1555        /// relation name
1556        name: ObjectName,
1557        kind: DescribeKind,
1558    },
1559    /// DESCRIBE FRAGMENT <fragment_id>
1560    DescribeFragment {
1561        fragment_id: u32,
1562    },
1563    /// SHOW OBJECT COMMAND
1564    ShowObjects {
1565        object: ShowObject,
1566        filter: Option<ShowStatementFilter>,
1567    },
1568    /// SHOW CREATE COMMAND
1569    ShowCreateObject {
1570        /// Show create object type
1571        create_type: ShowCreateType,
1572        /// Show create object name
1573        name: ObjectName,
1574    },
1575    ShowTransactionIsolationLevel,
1576    /// CANCEL JOBS COMMAND
1577    CancelJobs(JobIdents),
1578    /// KILL COMMAND
1579    /// Kill process in the show processlist.
1580    Kill(String),
1581    /// DROP
1582    Drop(DropStatement),
1583    /// DROP FUNCTION
1584    DropFunction {
1585        if_exists: bool,
1586        /// One or more function to drop
1587        func_desc: Vec<FunctionDesc>,
1588        /// `CASCADE` or `RESTRICT`
1589        option: Option<ReferentialAction>,
1590    },
1591    /// DROP AGGREGATE
1592    DropAggregate {
1593        if_exists: bool,
1594        /// One or more function to drop
1595        func_desc: Vec<FunctionDesc>,
1596        /// `CASCADE` or `RESTRICT`
1597        option: Option<ReferentialAction>,
1598    },
1599    /// `SET <variable>`
1600    ///
1601    /// Note: this is not a standard SQL statement, but it is supported by at
1602    /// least MySQL and PostgreSQL. Not all MySQL-specific syntactic forms are
1603    /// supported yet.
1604    SetVariable {
1605        local: bool,
1606        variable: Ident,
1607        value: SetVariableValue,
1608    },
1609    /// `SHOW <variable>`
1610    ///
1611    /// Note: this is a PostgreSQL-specific statement.
1612    ShowVariable {
1613        variable: Vec<Ident>,
1614    },
1615    /// `START TRANSACTION ...`
1616    StartTransaction {
1617        modes: Vec<TransactionMode>,
1618    },
1619    /// `BEGIN [ TRANSACTION | WORK ]`
1620    Begin {
1621        modes: Vec<TransactionMode>,
1622    },
1623    /// ABORT
1624    Abort,
1625    /// `SET TRANSACTION ...`
1626    SetTransaction {
1627        modes: Vec<TransactionMode>,
1628        snapshot: Option<Value>,
1629        session: bool,
1630    },
1631    /// `SET [ SESSION | LOCAL ] TIME ZONE { value | 'value' | LOCAL | DEFAULT }`
1632    SetTimeZone {
1633        local: bool,
1634        value: SetTimeZoneValue,
1635    },
1636    /// `COMMENT ON ...`
1637    ///
1638    /// Note: this is a PostgreSQL-specific statement.
1639    Comment {
1640        object_type: CommentObject,
1641        object_name: ObjectName,
1642        comment: Option<String>,
1643    },
1644    /// `COMMIT [ TRANSACTION | WORK ] [ AND [ NO ] CHAIN ]`
1645    Commit {
1646        chain: bool,
1647    },
1648    /// `ROLLBACK [ TRANSACTION | WORK ] [ AND [ NO ] CHAIN ]`
1649    Rollback {
1650        chain: bool,
1651    },
1652    /// CREATE SCHEMA
1653    CreateSchema {
1654        schema_name: ObjectName,
1655        if_not_exists: bool,
1656        owner: Option<ObjectName>,
1657    },
1658    /// CREATE DATABASE
1659    CreateDatabase {
1660        db_name: ObjectName,
1661        if_not_exists: bool,
1662        owner: Option<ObjectName>,
1663        resource_group: Option<SetVariableValue>,
1664        barrier_interval_ms: Option<u32>,
1665        checkpoint_frequency: Option<u64>,
1666    },
1667    /// GRANT privileges ON objects TO grantees
1668    Grant {
1669        privileges: Privileges,
1670        objects: GrantObjects,
1671        grantees: Vec<Ident>,
1672        with_grant_option: bool,
1673        granted_by: Option<Ident>,
1674    },
1675    /// REVOKE privileges ON objects FROM grantees
1676    Revoke {
1677        privileges: Privileges,
1678        objects: GrantObjects,
1679        grantees: Vec<Ident>,
1680        granted_by: Option<Ident>,
1681        revoke_grant_option: bool,
1682        cascade: bool,
1683    },
1684    /// `DEALLOCATE [ PREPARE ] { name | ALL }`
1685    ///
1686    /// Note: this is a PostgreSQL-specific statement.
1687    Deallocate {
1688        name: Option<Ident>,
1689        prepare: bool,
1690    },
1691    /// `EXECUTE name [ ( parameter [, ...] ) ]`
1692    ///
1693    /// Note: this is a PostgreSQL-specific statement.
1694    Execute {
1695        name: Ident,
1696        parameters: Vec<Expr>,
1697    },
1698    /// `PREPARE name [ ( data_type [, ...] ) ] AS statement`
1699    ///
1700    /// Note: this is a PostgreSQL-specific statement.
1701    Prepare {
1702        name: Ident,
1703        data_types: Vec<DataType>,
1704        statement: Box<Statement>,
1705    },
1706    /// EXPLAIN / DESCRIBE for select_statement
1707    Explain {
1708        /// Carry out the command and show actual run times and other statistics.
1709        analyze: bool,
1710        /// A SQL query that specifies what to explain
1711        statement: Box<Statement>,
1712        /// options of the explain statement
1713        options: ExplainOptions,
1714    },
1715    /// EXPLAIN ANALYZE for stream job
1716    /// We introduce a new statement rather than reuse `EXPLAIN` because
1717    /// the body of the statement is not an SQL query.
1718    /// TODO(kwannoel): Make profiling duration configurable: EXPLAIN ANALYZE (DURATION 1s) ...
1719    ExplainAnalyzeStreamJob {
1720        target: AnalyzeTarget,
1721        duration_secs: Option<u64>,
1722    },
1723    /// CREATE USER
1724    CreateUser(CreateUserStatement),
1725    /// ALTER USER
1726    AlterUser(AlterUserStatement),
1727    /// ALTER SYSTEM SET configuration_parameter { TO | = } { value | 'value' | DEFAULT }
1728    AlterSystem {
1729        param: Ident,
1730        value: SetVariableValue,
1731    },
1732    /// ALTER SYSTEM CLEAR FILE CACHE [META | DATA | ALL]
1733    AlterSystemClearFileCache {
1734        cache_type: FileCacheType,
1735    },
1736    /// FLUSH the current barrier.
1737    ///
1738    /// Note: RisingWave specific statement.
1739    Flush,
1740    /// WAIT for background stream jobs to finish.
1741    /// It will block the current session until the condition is met.
1742    Wait(WaitTarget),
1743    /// Trigger meta backup.
1744    Backup,
1745    /// Trigger stream job recover
1746    Recover,
1747    /// `USE <db_name>`
1748    ///
1749    /// Note: this is a RisingWave specific statement and used to switch the current database.
1750    Use {
1751        db_name: ObjectName,
1752    },
1753    /// `VACUUM [FULL | ORPHAN] [database_name][schema_name][object_name]`
1754    ///
1755    /// Note: this is a RisingWave specific statement for Iceberg table/sink maintenance.
1756    Vacuum {
1757        object_name: ObjectName,
1758        mode: VacuumMode,
1759    },
1760}
1761
1762#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
1763pub enum VacuumMode {
1764    Default,
1765    Full,
1766    OrphanFiles,
1767}
1768
1769#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1770pub enum DescribeKind {
1771    /// `DESCRIBE <name>`
1772    Plain,
1773
1774    /// `DESCRIBE FRAGMENTS <name>`
1775    Fragments,
1776}
1777
1778impl fmt::Display for Statement {
1779    /// Converts(unparses) the statement to a SQL string.
1780    ///
1781    /// If the resulting SQL is not valid, this function will panic. Use
1782    /// [`Statement::try_to_string`] to get a `Result` instead.
1783    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1784        // Note: we ignore formatting options here.
1785        let sql = self
1786            .try_to_string()
1787            .expect("normalized SQL should be parsable");
1788        f.write_str(&sql)
1789    }
1790}
1791
1792impl Statement {
1793    /// Converts(unparses) the statement to a SQL string.
1794    ///
1795    /// If the resulting SQL is not valid, returns an error.
1796    pub fn try_to_string(&self) -> Result<String, ParserError> {
1797        let sql = self.to_string_unchecked();
1798
1799        // TODO(#20713): expand this check to all statements
1800        if matches!(
1801            self,
1802            Statement::CreateTable { .. } | Statement::CreateSource { .. }
1803        ) {
1804            let _ = Parser::parse_sql(&sql)?;
1805        }
1806        Ok(sql)
1807    }
1808
1809    /// Converts(unparses) the statement to a SQL string.
1810    ///
1811    /// The result may not be valid SQL if there's an implementation bug in the `Display`
1812    /// trait of any AST node. To avoid this, always prefer [`Statement::try_to_string`]
1813    /// to get a `Result`, or `to_string` which panics if the SQL is invalid.
1814    pub fn to_string_unchecked(&self) -> String {
1815        let mut buf = String::new();
1816        self.fmt_unchecked(&mut buf).unwrap();
1817        buf
1818    }
1819
1820    // NOTE: This function should not check the validity of the unparsed SQL (and panic).
1821    //       Thus, do not directly format a statement with `write!` or `format!`. Recursively
1822    //       call `fmt_unchecked` on the inner statements instead.
1823    //
1824    // Clippy thinks this function is too complicated, but it is painful to
1825    // split up without extracting structs for each `Statement` variant.
1826    #[expect(clippy::cognitive_complexity)]
1827    fn fmt_unchecked(&self, mut f: impl std::fmt::Write) -> fmt::Result {
1828        match self {
1829            Statement::Explain {
1830                analyze,
1831                statement,
1832                options,
1833            } => {
1834                write!(f, "EXPLAIN ")?;
1835
1836                if *analyze {
1837                    write!(f, "ANALYZE ")?;
1838                }
1839                write!(f, "{}", options)?;
1840
1841                statement.fmt_unchecked(f)
1842            }
1843            Statement::ExplainAnalyzeStreamJob {
1844                target,
1845                duration_secs,
1846            } => {
1847                write!(f, "EXPLAIN ANALYZE {}", target)?;
1848                if let Some(duration_secs) = duration_secs {
1849                    write!(f, " (DURATION_SECS {})", duration_secs)?;
1850                }
1851                Ok(())
1852            }
1853            Statement::Query(s) => write!(f, "{}", s),
1854            Statement::Truncate { table_name } => {
1855                write!(f, "TRUNCATE TABLE {}", table_name)?;
1856                Ok(())
1857            }
1858            Statement::Refresh { table_name } => {
1859                write!(f, "REFRESH TABLE {}", table_name)?;
1860                Ok(())
1861            }
1862            Statement::Analyze { table_name } => {
1863                write!(f, "ANALYZE TABLE {}", table_name)?;
1864                Ok(())
1865            }
1866            Statement::Describe { name, kind } => {
1867                write!(f, "DESCRIBE {}", name)?;
1868                match kind {
1869                    DescribeKind::Plain => {}
1870
1871                    DescribeKind::Fragments => {
1872                        write!(f, " FRAGMENTS")?;
1873                    }
1874                }
1875                Ok(())
1876            }
1877            Statement::DescribeFragment { fragment_id } => {
1878                write!(f, "DESCRIBE FRAGMENT {}", fragment_id)?;
1879                Ok(())
1880            }
1881            Statement::ShowObjects {
1882                object: show_object,
1883                filter,
1884            } => {
1885                write!(f, "SHOW {}", show_object)?;
1886                if let Some(filter) = filter {
1887                    write!(f, " {}", filter)?;
1888                }
1889                Ok(())
1890            }
1891            Statement::ShowCreateObject {
1892                create_type: show_type,
1893                name,
1894            } => {
1895                write!(f, "SHOW CREATE {} {}", show_type, name)?;
1896                Ok(())
1897            }
1898            Statement::ShowTransactionIsolationLevel => {
1899                write!(f, "SHOW TRANSACTION ISOLATION LEVEL")?;
1900                Ok(())
1901            }
1902            Statement::Insert {
1903                table_name,
1904                columns,
1905                source,
1906                returning,
1907            } => {
1908                write!(f, "INSERT INTO {table_name} ", table_name = table_name,)?;
1909                if !columns.is_empty() {
1910                    write!(f, "({}) ", display_comma_separated(columns))?;
1911                }
1912                write!(f, "{}", source)?;
1913                if !returning.is_empty() {
1914                    write!(f, " RETURNING ({})", display_comma_separated(returning))?;
1915                }
1916                Ok(())
1917            }
1918            Statement::Copy { entity, target } => {
1919                write!(f, "COPY ",)?;
1920                match entity {
1921                    CopyEntity::Query(query) => {
1922                        write!(f, "({})", query)?;
1923                    }
1924                    CopyEntity::Table {
1925                        table_name,
1926                        columns,
1927                    } => {
1928                        write!(f, "{}", table_name)?;
1929                        if !columns.is_empty() {
1930                            write!(f, " ({})", display_comma_separated(columns))?;
1931                        }
1932                    }
1933                }
1934
1935                match target {
1936                    CopyTarget::Stdin { values } => {
1937                        write!(f, " FROM STDIN; ")?;
1938                        if !values.is_empty() {
1939                            writeln!(f)?;
1940                            let mut delim = "";
1941                            for v in values {
1942                                write!(f, "{}", delim)?;
1943                                delim = "\t";
1944                                if let Some(v) = v {
1945                                    write!(f, "{}", v)?;
1946                                } else {
1947                                    write!(f, "\\N")?;
1948                                }
1949                            }
1950                        }
1951                        write!(f, "\n\\.")
1952                    }
1953                    CopyTarget::Stdout => {
1954                        write!(f, " TO STDOUT")
1955                    }
1956                }
1957            }
1958            Statement::Update {
1959                table_name,
1960                assignments,
1961                selection,
1962                returning,
1963            } => {
1964                write!(f, "UPDATE {}", table_name)?;
1965                if !assignments.is_empty() {
1966                    write!(f, " SET {}", display_comma_separated(assignments))?;
1967                }
1968                if let Some(selection) = selection {
1969                    write!(f, " WHERE {}", selection)?;
1970                }
1971                if !returning.is_empty() {
1972                    write!(f, " RETURNING ({})", display_comma_separated(returning))?;
1973                }
1974                Ok(())
1975            }
1976            Statement::Delete {
1977                table_name,
1978                selection,
1979                returning,
1980            } => {
1981                write!(f, "DELETE FROM {}", table_name)?;
1982                if let Some(selection) = selection {
1983                    write!(f, " WHERE {}", selection)?;
1984                }
1985                if !returning.is_empty() {
1986                    write!(f, " RETURNING {}", display_comma_separated(returning))?;
1987                }
1988                Ok(())
1989            }
1990            Statement::DeleteMetaSnapshots { snapshot_ids } => {
1991                write!(
1992                    f,
1993                    "DELETE META SNAPSHOTS {}",
1994                    display_comma_separated(snapshot_ids)
1995                )?;
1996                Ok(())
1997            }
1998            Statement::CreateDatabase {
1999                db_name,
2000                if_not_exists,
2001                owner,
2002                resource_group,
2003                barrier_interval_ms,
2004                checkpoint_frequency,
2005            } => {
2006                write!(f, "CREATE DATABASE")?;
2007                if *if_not_exists {
2008                    write!(f, " IF NOT EXISTS")?;
2009                }
2010                write!(f, " {}", db_name)?;
2011                if let Some(owner) = owner {
2012                    write!(f, " WITH OWNER = {}", owner)?;
2013                }
2014                if let Some(resource_group) = resource_group {
2015                    write!(f, " RESOURCE_GROUP = {}", resource_group)?;
2016                }
2017                if let Some(barrier_interval_ms) = barrier_interval_ms {
2018                    write!(f, " BARRIER_INTERVAL_MS = {}", barrier_interval_ms)?;
2019                }
2020                if let Some(checkpoint_frequency) = checkpoint_frequency {
2021                    write!(f, " CHECKPOINT_FREQUENCY = {}", checkpoint_frequency)?;
2022                }
2023
2024                Ok(())
2025            }
2026            Statement::CreateFunction {
2027                or_replace,
2028                temporary,
2029                if_not_exists,
2030                name,
2031                args,
2032                returns,
2033                params,
2034                with_options,
2035            } => {
2036                write!(
2037                    f,
2038                    "CREATE {or_replace}{temp}FUNCTION {if_not_exists}{name}",
2039                    temp = if *temporary { "TEMPORARY " } else { "" },
2040                    or_replace = if *or_replace { "OR REPLACE " } else { "" },
2041                    if_not_exists = if *if_not_exists { "IF NOT EXISTS " } else { "" },
2042                )?;
2043                if let Some(args) = args {
2044                    write!(f, "({})", display_comma_separated(args))?;
2045                }
2046                if let Some(return_type) = returns {
2047                    write!(f, " {}", return_type)?;
2048                }
2049                write!(f, "{params}")?;
2050                write!(f, "{with_options}")?;
2051                Ok(())
2052            }
2053            Statement::CreateAggregate {
2054                or_replace,
2055                if_not_exists,
2056                name,
2057                args,
2058                returns,
2059                append_only,
2060                params,
2061            } => {
2062                write!(
2063                    f,
2064                    "CREATE {or_replace}AGGREGATE {if_not_exists}{name}",
2065                    or_replace = if *or_replace { "OR REPLACE " } else { "" },
2066                    if_not_exists = if *if_not_exists { "IF NOT EXISTS " } else { "" },
2067                )?;
2068                write!(f, "({})", display_comma_separated(args))?;
2069                write!(f, " RETURNS {}", returns)?;
2070                if *append_only {
2071                    write!(f, " APPEND ONLY")?;
2072                }
2073                write!(f, "{params}")?;
2074                Ok(())
2075            }
2076            Statement::CreateView {
2077                name,
2078                or_replace,
2079                if_not_exists,
2080                columns,
2081                query,
2082                materialized,
2083                with_options,
2084                emit_mode,
2085            } => {
2086                write!(
2087                    f,
2088                    "CREATE {or_replace}{materialized}VIEW {if_not_exists}{name}",
2089                    or_replace = if *or_replace { "OR REPLACE " } else { "" },
2090                    materialized = if *materialized { "MATERIALIZED " } else { "" },
2091                    if_not_exists = if *if_not_exists { "IF NOT EXISTS " } else { "" },
2092                    name = name
2093                )?;
2094                if !with_options.is_empty() {
2095                    write!(f, " WITH ({})", display_comma_separated(with_options))?;
2096                }
2097                if !columns.is_empty() {
2098                    write!(f, " ({})", display_comma_separated(columns))?;
2099                }
2100                write!(f, " AS {}", query)?;
2101                if let Some(emit_mode) = emit_mode {
2102                    write!(f, " EMIT {}", emit_mode)?;
2103                }
2104                Ok(())
2105            }
2106            Statement::CreateTable {
2107                name,
2108                columns,
2109                wildcard_idx,
2110                constraints,
2111                with_options,
2112                or_replace,
2113                if_not_exists,
2114                temporary,
2115                format_encode,
2116                source_watermarks,
2117                append_only,
2118                on_conflict,
2119                with_version_columns,
2120                query,
2121                cdc_table_info,
2122                include_column_options,
2123                webhook_info,
2124                engine,
2125            } => {
2126                // We want to allow the following options
2127                // Empty column list, allowed by PostgreSQL:
2128                //   `CREATE TABLE t ()`
2129                // No columns provided for CREATE TABLE AS:
2130                //   `CREATE TABLE t AS SELECT a from t2`
2131                // Columns provided for CREATE TABLE AS:
2132                //   `CREATE TABLE t (a INT) AS SELECT a from t2`
2133                write!(
2134                    f,
2135                    "CREATE {or_replace}{temporary}TABLE {if_not_exists}{name}",
2136                    or_replace = if *or_replace { "OR REPLACE " } else { "" },
2137                    if_not_exists = if *if_not_exists { "IF NOT EXISTS " } else { "" },
2138                    temporary = if *temporary { "TEMPORARY " } else { "" },
2139                    name = name,
2140                )?;
2141                if !columns.is_empty() || !constraints.is_empty() {
2142                    write!(
2143                        f,
2144                        " {}",
2145                        fmt_create_items(columns, constraints, source_watermarks, *wildcard_idx)?
2146                    )?;
2147                } else if query.is_none() {
2148                    // PostgreSQL allows `CREATE TABLE t ();`, but requires empty parens
2149                    write!(f, " ()")?;
2150                }
2151                if *append_only {
2152                    write!(f, " APPEND ONLY")?;
2153                }
2154
2155                if let Some(on_conflict_behavior) = on_conflict {
2156                    write!(f, " ON CONFLICT {}", on_conflict_behavior)?;
2157                }
2158                if !with_version_columns.is_empty() {
2159                    write!(
2160                        f,
2161                        " WITH VERSION COLUMN({})",
2162                        display_comma_separated(with_version_columns)
2163                    )?;
2164                }
2165                if !include_column_options.is_empty() {
2166                    write!(f, " {}", display_separated(include_column_options, " "))?;
2167                }
2168                if !with_options.is_empty() {
2169                    write!(f, " WITH ({})", display_comma_separated(with_options))?;
2170                }
2171                if let Some(format_encode) = format_encode {
2172                    write!(f, " {}", format_encode)?;
2173                }
2174                if let Some(query) = query {
2175                    write!(f, " AS {}", query)?;
2176                }
2177                if let Some(info) = cdc_table_info {
2178                    write!(f, " FROM {}", info.source_name)?;
2179                    write!(
2180                        f,
2181                        " TABLE '{}'",
2182                        value::escape_single_quote_string(&info.external_table_name)
2183                    )?;
2184                }
2185                if let Some(info) = webhook_info
2186                    && let Some(signature_expr) = &info.signature_expr
2187                {
2188                    if let Some(secret) = &info.secret_ref {
2189                        write!(f, " VALIDATE SECRET {}", secret.secret_name)?;
2190                    } else {
2191                        write!(f, " VALIDATE")?;
2192                    }
2193                    write!(f, " AS {}", signature_expr)?;
2194                }
2195                match engine {
2196                    Engine::Hummock => {}
2197                    Engine::Iceberg => {
2198                        write!(f, " ENGINE = {}", engine)?;
2199                    }
2200                }
2201                Ok(())
2202            }
2203            Statement::CreateIndex {
2204                name,
2205                table_name,
2206                columns,
2207                method,
2208                include,
2209                distributed_by,
2210                unique,
2211                if_not_exists,
2212                with_properties,
2213            } => write!(
2214                f,
2215                "CREATE {unique}INDEX {if_not_exists}{name} ON {table_name}{method}({columns}){include}{distributed_by}{with_properties}",
2216                unique = if *unique { "UNIQUE " } else { "" },
2217                if_not_exists = if *if_not_exists { "IF NOT EXISTS " } else { "" },
2218                name = name,
2219                table_name = table_name,
2220                method = if let Some(method) = method {
2221                    format!(" USING {} ", method)
2222                } else {
2223                    "".to_owned()
2224                },
2225                columns = display_comma_separated(columns),
2226                include = if include.is_empty() {
2227                    "".to_owned()
2228                } else {
2229                    format!(" INCLUDE({})", display_separated(include, ","))
2230                },
2231                distributed_by = if distributed_by.is_empty() {
2232                    "".to_owned()
2233                } else {
2234                    format!(
2235                        " DISTRIBUTED BY({})",
2236                        display_separated(distributed_by, ",")
2237                    )
2238                },
2239                with_properties = if !with_properties.0.is_empty() {
2240                    format!(" {}", with_properties)
2241                } else {
2242                    "".to_owned()
2243                },
2244            ),
2245            Statement::CreateSource { stmt } => write!(f, "CREATE SOURCE {}", stmt,),
2246            Statement::CreateSink { stmt } => {
2247                if stmt.or_replace {
2248                    write!(f, "REPLACE SINK {}", stmt)
2249                } else {
2250                    write!(f, "CREATE SINK {}", stmt)
2251                }
2252            }
2253            Statement::CreateSubscription { stmt } => write!(f, "CREATE SUBSCRIPTION {}", stmt,),
2254            Statement::CreateConnection { stmt } => write!(f, "CREATE CONNECTION {}", stmt,),
2255            Statement::DeclareCursor { stmt } => write!(f, "DECLARE {}", stmt,),
2256            Statement::FetchCursor { stmt } => write!(f, "FETCH {}", stmt),
2257            Statement::CloseCursor { stmt } => write!(f, "CLOSE {}", stmt),
2258            Statement::CreateSecret { stmt } => write!(f, "CREATE SECRET {}", stmt),
2259            Statement::AlterDatabase { name, operation } => {
2260                write!(f, "ALTER DATABASE {} {}", name, operation)
2261            }
2262            Statement::AlterSchema { name, operation } => {
2263                write!(f, "ALTER SCHEMA {} {}", name, operation)
2264            }
2265            Statement::AlterTable { name, operation } => {
2266                write!(f, "ALTER TABLE {} {}", name, operation)
2267            }
2268            Statement::AlterIndex { name, operation } => {
2269                write!(f, "ALTER INDEX {} {}", name, operation)
2270            }
2271            Statement::AlterView {
2272                materialized,
2273                name,
2274                operation,
2275            } => {
2276                write!(
2277                    f,
2278                    "ALTER {}VIEW {} {}",
2279                    if *materialized { "MATERIALIZED " } else { "" },
2280                    name,
2281                    operation
2282                )
2283            }
2284            Statement::AlterSink { name, operation } => {
2285                write!(f, "ALTER SINK {} {}", name, operation)
2286            }
2287            Statement::AlterSubscription { name, operation } => {
2288                write!(f, "ALTER SUBSCRIPTION {} {}", name, operation)
2289            }
2290            Statement::AlterSource { name, operation } => {
2291                write!(f, "ALTER SOURCE {} {}", name, operation)
2292            }
2293            Statement::AlterFunction {
2294                name,
2295                args,
2296                operation,
2297            } => {
2298                write!(f, "ALTER FUNCTION {}", name)?;
2299                if let Some(args) = args {
2300                    write!(f, "({})", display_comma_separated(args))?;
2301                }
2302                write!(f, " {}", operation)
2303            }
2304            Statement::AlterConnection { name, operation } => {
2305                write!(f, "ALTER CONNECTION {} {}", name, operation)
2306            }
2307            Statement::AlterSecret { name, operation } => {
2308                write!(f, "ALTER SECRET {}", name)?;
2309                write!(f, "{}", operation)
2310            }
2311            Statement::Discard(t) => write!(f, "DISCARD {}", t),
2312            Statement::Drop(stmt) => write!(f, "DROP {}", stmt),
2313            Statement::DropFunction {
2314                if_exists,
2315                func_desc,
2316                option,
2317            } => {
2318                write!(
2319                    f,
2320                    "DROP FUNCTION{} {}",
2321                    if *if_exists { " IF EXISTS" } else { "" },
2322                    display_comma_separated(func_desc),
2323                )?;
2324                if let Some(op) = option {
2325                    write!(f, " {}", op)?;
2326                }
2327                Ok(())
2328            }
2329            Statement::DropAggregate {
2330                if_exists,
2331                func_desc,
2332                option,
2333            } => {
2334                write!(
2335                    f,
2336                    "DROP AGGREGATE{} {}",
2337                    if *if_exists { " IF EXISTS" } else { "" },
2338                    display_comma_separated(func_desc),
2339                )?;
2340                if let Some(op) = option {
2341                    write!(f, " {}", op)?;
2342                }
2343                Ok(())
2344            }
2345            Statement::SetVariable {
2346                local,
2347                variable,
2348                value,
2349            } => {
2350                f.write_str("SET ")?;
2351                if *local {
2352                    f.write_str("LOCAL ")?;
2353                }
2354                write!(f, "{name} = {value}", name = variable,)
2355            }
2356            Statement::ShowVariable { variable } => {
2357                write!(f, "SHOW")?;
2358                if !variable.is_empty() {
2359                    write!(f, " {}", display_separated(variable, " "))?;
2360                }
2361                Ok(())
2362            }
2363            Statement::StartTransaction { modes } => {
2364                write!(f, "START TRANSACTION")?;
2365                if !modes.is_empty() {
2366                    write!(f, " {}", display_comma_separated(modes))?;
2367                }
2368                Ok(())
2369            }
2370            Statement::Abort => {
2371                write!(f, "ABORT")?;
2372                Ok(())
2373            }
2374            Statement::SetTransaction {
2375                modes,
2376                snapshot,
2377                session,
2378            } => {
2379                if *session {
2380                    write!(f, "SET SESSION CHARACTERISTICS AS TRANSACTION")?;
2381                } else {
2382                    write!(f, "SET TRANSACTION")?;
2383                }
2384                if !modes.is_empty() {
2385                    write!(f, " {}", display_comma_separated(modes))?;
2386                }
2387                if let Some(snapshot_id) = snapshot {
2388                    write!(f, " SNAPSHOT {}", snapshot_id)?;
2389                }
2390                Ok(())
2391            }
2392            Statement::SetTimeZone { local, value } => {
2393                write!(f, "SET")?;
2394                if *local {
2395                    write!(f, " LOCAL")?;
2396                }
2397                write!(f, " TIME ZONE {}", value)?;
2398                Ok(())
2399            }
2400            Statement::Commit { chain } => {
2401                write!(f, "COMMIT{}", if *chain { " AND CHAIN" } else { "" },)
2402            }
2403            Statement::Rollback { chain } => {
2404                write!(f, "ROLLBACK{}", if *chain { " AND CHAIN" } else { "" },)
2405            }
2406            Statement::CreateSchema {
2407                schema_name,
2408                if_not_exists,
2409                owner,
2410            } => {
2411                write!(
2412                    f,
2413                    "CREATE SCHEMA {if_not_exists}{name}",
2414                    if_not_exists = if *if_not_exists { "IF NOT EXISTS " } else { "" },
2415                    name = schema_name
2416                )?;
2417                if let Some(user) = owner {
2418                    write!(f, " AUTHORIZATION {}", user)?;
2419                }
2420                Ok(())
2421            }
2422            Statement::Grant {
2423                privileges,
2424                objects,
2425                grantees,
2426                with_grant_option,
2427                granted_by,
2428            } => {
2429                write!(f, "GRANT {} ", privileges)?;
2430                write!(f, "ON {} ", objects)?;
2431                write!(f, "TO {}", display_comma_separated(grantees))?;
2432                if *with_grant_option {
2433                    write!(f, " WITH GRANT OPTION")?;
2434                }
2435                if let Some(grantor) = granted_by {
2436                    write!(f, " GRANTED BY {}", grantor)?;
2437                }
2438                Ok(())
2439            }
2440            Statement::Revoke {
2441                privileges,
2442                objects,
2443                grantees,
2444                granted_by,
2445                revoke_grant_option,
2446                cascade,
2447            } => {
2448                write!(
2449                    f,
2450                    "REVOKE {}{} ",
2451                    if *revoke_grant_option {
2452                        "GRANT OPTION FOR "
2453                    } else {
2454                        ""
2455                    },
2456                    privileges
2457                )?;
2458                write!(f, "ON {} ", objects)?;
2459                write!(f, "FROM {}", display_comma_separated(grantees))?;
2460                if let Some(grantor) = granted_by {
2461                    write!(f, " GRANTED BY {}", grantor)?;
2462                }
2463                write!(f, " {}", if *cascade { "CASCADE" } else { "RESTRICT" })?;
2464                Ok(())
2465            }
2466            Statement::Deallocate { name, prepare } => {
2467                if let Some(name) = name {
2468                    write!(
2469                        f,
2470                        "DEALLOCATE {prepare}{name}",
2471                        prepare = if *prepare { "PREPARE " } else { "" },
2472                        name = name,
2473                    )
2474                } else {
2475                    write!(
2476                        f,
2477                        "DEALLOCATE {prepare}ALL",
2478                        prepare = if *prepare { "PREPARE " } else { "" },
2479                    )
2480                }
2481            }
2482            Statement::Execute { name, parameters } => {
2483                write!(f, "EXECUTE {}", name)?;
2484                if !parameters.is_empty() {
2485                    write!(f, "({})", display_comma_separated(parameters))?;
2486                }
2487                Ok(())
2488            }
2489            Statement::Prepare {
2490                name,
2491                data_types,
2492                statement,
2493            } => {
2494                write!(f, "PREPARE {} ", name)?;
2495                if !data_types.is_empty() {
2496                    write!(f, "({}) ", display_comma_separated(data_types))?;
2497                }
2498                write!(f, "AS ")?;
2499                statement.fmt_unchecked(f)
2500            }
2501            Statement::Comment {
2502                object_type,
2503                object_name,
2504                comment,
2505            } => {
2506                write!(f, "COMMENT ON {} {} IS ", object_type, object_name)?;
2507                if let Some(c) = comment {
2508                    write!(f, "'{}'", c)
2509                } else {
2510                    write!(f, "NULL")
2511                }
2512            }
2513            Statement::CreateUser(statement) => {
2514                write!(f, "CREATE USER {}", statement)
2515            }
2516            Statement::AlterUser(statement) => {
2517                write!(f, "ALTER USER {}", statement)
2518            }
2519            Statement::AlterSystem { param, value } => {
2520                f.write_str("ALTER SYSTEM SET ")?;
2521                write!(f, "{param} = {value}",)
2522            }
2523            Statement::AlterSystemClearFileCache { cache_type } => {
2524                f.write_str("ALTER SYSTEM CLEAR FILE CACHE ")?;
2525                match cache_type {
2526                    FileCacheType::Meta => f.write_str("META"),
2527                    FileCacheType::Data => f.write_str("DATA"),
2528                    FileCacheType::All => f.write_str("ALL"),
2529                }
2530            }
2531            Statement::Flush => {
2532                write!(f, "FLUSH")
2533            }
2534            Statement::Wait(target) => match target {
2535                WaitTarget::All => write!(f, "WAIT"),
2536                WaitTarget::Table(name) => write!(f, "WAIT TABLE {name}"),
2537                WaitTarget::MaterializedView(name) => {
2538                    write!(f, "WAIT MATERIALIZED VIEW {name}")
2539                }
2540                WaitTarget::Sink(name) => write!(f, "WAIT SINK {name}"),
2541                WaitTarget::Index(name) => write!(f, "WAIT INDEX {name}"),
2542            },
2543            Statement::Backup => {
2544                write!(f, "BACKUP")?;
2545                Ok(())
2546            }
2547            Statement::Begin { modes } => {
2548                write!(f, "BEGIN")?;
2549                if !modes.is_empty() {
2550                    write!(f, " {}", display_comma_separated(modes))?;
2551                }
2552                Ok(())
2553            }
2554            Statement::CancelJobs(jobs) => {
2555                write!(f, "CANCEL JOBS {}", display_comma_separated(&jobs.0))?;
2556                Ok(())
2557            }
2558            Statement::Kill(worker_process_id) => {
2559                write!(f, "KILL '{}'", worker_process_id)?;
2560                Ok(())
2561            }
2562            Statement::Recover => {
2563                write!(f, "RECOVER")?;
2564                Ok(())
2565            }
2566            Statement::Use { db_name } => {
2567                write!(f, "USE {}", db_name)?;
2568                Ok(())
2569            }
2570            Statement::Vacuum { object_name, mode } => match mode {
2571                VacuumMode::Default => write!(f, "VACUUM {}", object_name),
2572                VacuumMode::Full => write!(f, "VACUUM FULL {}", object_name),
2573                VacuumMode::OrphanFiles => write!(f, "VACUUM ORPHAN {}", object_name),
2574            },
2575            Statement::AlterFragment {
2576                fragment_ids,
2577                operation,
2578            } => {
2579                write!(
2580                    f,
2581                    "ALTER FRAGMENT {} {}",
2582                    display_comma_separated(fragment_ids),
2583                    operation
2584                )
2585            }
2586            Statement::AlterCompactionGroup {
2587                group_ids,
2588                operation,
2589            } => {
2590                write!(
2591                    f,
2592                    "ALTER COMPACTION GROUP {} {}",
2593                    display_comma_separated(group_ids),
2594                    operation
2595                )
2596            }
2597            Statement::AlterDefaultPrivileges {
2598                target_users,
2599                schema_names,
2600                operation,
2601            } => {
2602                write!(f, "ALTER DEFAULT PRIVILEGES")?;
2603                if let Some(target_users) = target_users {
2604                    write!(f, " FOR {}", display_comma_separated(target_users))?;
2605                }
2606                if let Some(schema_names) = schema_names {
2607                    write!(f, " IN SCHEMA {}", display_comma_separated(schema_names))?;
2608                }
2609                write!(f, " {}", operation)
2610            }
2611        }
2612    }
2613
2614    pub fn is_create(&self) -> bool {
2615        matches!(
2616            self,
2617            Statement::CreateTable { .. }
2618                | Statement::CreateView { .. }
2619                | Statement::CreateSource { .. }
2620                | Statement::CreateSink { .. }
2621                | Statement::CreateSubscription { .. }
2622                | Statement::CreateConnection { .. }
2623                | Statement::CreateSecret { .. }
2624                | Statement::CreateUser { .. }
2625                | Statement::CreateDatabase { .. }
2626                | Statement::CreateFunction { .. }
2627                | Statement::CreateAggregate { .. }
2628                | Statement::CreateIndex { .. }
2629                | Statement::CreateSchema { .. }
2630        )
2631    }
2632}
2633
2634impl Display for IncludeOptionItem {
2635    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2636        let Self {
2637            column_type,
2638            inner_field,
2639            header_inner_expect_type,
2640            column_alias,
2641        } = self;
2642        write!(f, "INCLUDE {}", column_type)?;
2643        if let Some(inner_field) = inner_field {
2644            write!(f, " '{}'", value::escape_single_quote_string(inner_field))?;
2645            if let Some(expected_type) = header_inner_expect_type {
2646                write!(f, " {}", expected_type)?;
2647            }
2648        }
2649        if let Some(alias) = column_alias {
2650            write!(f, " AS {}", alias)?;
2651        }
2652        Ok(())
2653    }
2654}
2655
2656#[derive(Debug, Clone, PartialEq, Eq, Hash)]
2657#[non_exhaustive]
2658pub enum OnInsert {
2659    /// ON DUPLICATE KEY UPDATE (MySQL when the key already exists, then execute an update instead)
2660    DuplicateKeyUpdate(Vec<Assignment>),
2661}
2662
2663impl fmt::Display for OnInsert {
2664    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2665        match self {
2666            Self::DuplicateKeyUpdate(expr) => write!(
2667                f,
2668                " ON DUPLICATE KEY UPDATE {}",
2669                display_comma_separated(expr)
2670            ),
2671        }
2672    }
2673}
2674
2675/// Privileges granted in a GRANT statement or revoked in a REVOKE statement.
2676#[derive(Debug, Clone, PartialEq, Eq, Hash)]
2677pub enum Privileges {
2678    /// All privileges applicable to the object type
2679    All {
2680        /// Optional keyword from the spec, ignored in practice
2681        with_privileges_keyword: bool,
2682    },
2683    /// Specific privileges (e.g. `SELECT`, `INSERT`)
2684    Actions(Vec<Action>),
2685}
2686
2687impl fmt::Display for Privileges {
2688    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2689        match self {
2690            Privileges::All {
2691                with_privileges_keyword,
2692            } => {
2693                write!(
2694                    f,
2695                    "ALL{}",
2696                    if *with_privileges_keyword {
2697                        " PRIVILEGES"
2698                    } else {
2699                        ""
2700                    }
2701                )
2702            }
2703            Privileges::Actions(actions) => {
2704                write!(f, "{}", display_comma_separated(actions))
2705            }
2706        }
2707    }
2708}
2709
2710/// A privilege on a database object (table, sequence, etc.).
2711#[derive(Debug, Clone, PartialEq, Eq, Hash)]
2712pub enum Action {
2713    Connect,
2714    Create,
2715    Delete,
2716    Execute,
2717    Insert { columns: Option<Vec<Ident>> },
2718    References { columns: Option<Vec<Ident>> },
2719    Select { columns: Option<Vec<Ident>> },
2720    Temporary,
2721    Trigger,
2722    Truncate,
2723    Update { columns: Option<Vec<Ident>> },
2724    Usage,
2725}
2726
2727impl fmt::Display for Action {
2728    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2729        match self {
2730            Action::Connect => f.write_str("CONNECT")?,
2731            Action::Create => f.write_str("CREATE")?,
2732            Action::Delete => f.write_str("DELETE")?,
2733            Action::Execute => f.write_str("EXECUTE")?,
2734            Action::Insert { .. } => f.write_str("INSERT")?,
2735            Action::References { .. } => f.write_str("REFERENCES")?,
2736            Action::Select { .. } => f.write_str("SELECT")?,
2737            Action::Temporary => f.write_str("TEMPORARY")?,
2738            Action::Trigger => f.write_str("TRIGGER")?,
2739            Action::Truncate => f.write_str("TRUNCATE")?,
2740            Action::Update { .. } => f.write_str("UPDATE")?,
2741            Action::Usage => f.write_str("USAGE")?,
2742        };
2743        match self {
2744            Action::Insert { columns }
2745            | Action::References { columns }
2746            | Action::Select { columns }
2747            | Action::Update { columns } => {
2748                if let Some(columns) = columns {
2749                    write!(f, " ({})", display_comma_separated(columns))?;
2750                }
2751            }
2752            _ => (),
2753        };
2754        Ok(())
2755    }
2756}
2757
2758/// Objects on which privileges are granted in a GRANT statement.
2759#[derive(Debug, Clone, PartialEq, Eq, Hash)]
2760pub enum GrantObjects {
2761    /// Grant privileges on `ALL SEQUENCES IN SCHEMA <schema_name> [, ...]`
2762    AllSequencesInSchema { schemas: Vec<ObjectName> },
2763    /// Grant privileges on `ALL TABLES IN SCHEMA <schema_name> [, ...]`
2764    AllTablesInSchema { schemas: Vec<ObjectName> },
2765    /// Grant privileges on `ALL SOURCES IN SCHEMA <schema_name> [, ...]`
2766    AllSourcesInSchema { schemas: Vec<ObjectName> },
2767    /// Grant privileges on `ALL SINKS IN SCHEMA <schema_name> [, ...]`
2768    AllSinksInSchema { schemas: Vec<ObjectName> },
2769    /// Grant privileges on `ALL MATERIALIZED VIEWS IN SCHEMA <schema_name> [, ...]`
2770    AllMviewsInSchema { schemas: Vec<ObjectName> },
2771    /// Grant privileges on `ALL VIEWS IN SCHEMA <schema_name> [, ...]`
2772    AllViewsInSchema { schemas: Vec<ObjectName> },
2773    /// Grant privileges on `ALL FUNCTIONS IN SCHEMA <schema_name> [, ...]`
2774    AllFunctionsInSchema { schemas: Vec<ObjectName> },
2775    /// Grant privileges on `ALL SECRETS IN SCHEMA <schema_name> [, ...]`
2776    AllSecretsInSchema { schemas: Vec<ObjectName> },
2777    /// Grant privileges on `ALL SUBSCRIPTIONS IN SCHEMA <schema_name> [, ...]`
2778    AllSubscriptionsInSchema { schemas: Vec<ObjectName> },
2779    /// Grant privileges on `ALL CONNECTIONS IN SCHEMA <schema_name> [, ...]`
2780    AllConnectionsInSchema { schemas: Vec<ObjectName> },
2781    /// Grant privileges on specific databases
2782    Databases(Vec<ObjectName>),
2783    /// Grant privileges on specific schemas
2784    Schemas(Vec<ObjectName>),
2785    /// Grant privileges on specific sources
2786    Sources(Vec<ObjectName>),
2787    /// Grant privileges on specific materialized views
2788    Mviews(Vec<ObjectName>),
2789    /// Grant privileges on specific sequences
2790    Sequences(Vec<ObjectName>),
2791    /// Grant privileges on specific tables
2792    Tables(Vec<ObjectName>),
2793    /// Grant privileges on specific sinks
2794    Sinks(Vec<ObjectName>),
2795    /// Grant privileges on specific views
2796    Views(Vec<ObjectName>),
2797    /// Grant privileges on specific connections
2798    Connections(Vec<ObjectName>),
2799    /// Grant privileges on specific subscriptions
2800    Subscriptions(Vec<ObjectName>),
2801    /// Grant privileges on specific functions
2802    Functions(Vec<FunctionDesc>),
2803    /// Grant privileges on specific secrets
2804    Secrets(Vec<ObjectName>),
2805}
2806
2807impl fmt::Display for GrantObjects {
2808    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2809        match self {
2810            GrantObjects::Sequences(sequences) => {
2811                write!(f, "SEQUENCE {}", display_comma_separated(sequences))
2812            }
2813            GrantObjects::Schemas(schemas) => {
2814                write!(f, "SCHEMA {}", display_comma_separated(schemas))
2815            }
2816            GrantObjects::Tables(tables) => {
2817                write!(f, "{}", display_comma_separated(tables))
2818            }
2819            GrantObjects::AllSequencesInSchema { schemas } => {
2820                write!(
2821                    f,
2822                    "ALL SEQUENCES IN SCHEMA {}",
2823                    display_comma_separated(schemas)
2824                )
2825            }
2826            GrantObjects::AllTablesInSchema { schemas } => {
2827                write!(
2828                    f,
2829                    "ALL TABLES IN SCHEMA {}",
2830                    display_comma_separated(schemas)
2831                )
2832            }
2833            GrantObjects::AllSourcesInSchema { schemas } => {
2834                write!(
2835                    f,
2836                    "ALL SOURCES IN SCHEMA {}",
2837                    display_comma_separated(schemas)
2838                )
2839            }
2840            GrantObjects::AllMviewsInSchema { schemas } => {
2841                write!(
2842                    f,
2843                    "ALL MATERIALIZED VIEWS IN SCHEMA {}",
2844                    display_comma_separated(schemas)
2845                )
2846            }
2847            GrantObjects::AllSinksInSchema { schemas } => {
2848                write!(
2849                    f,
2850                    "ALL SINKS IN SCHEMA {}",
2851                    display_comma_separated(schemas)
2852                )
2853            }
2854            GrantObjects::AllViewsInSchema { schemas } => {
2855                write!(
2856                    f,
2857                    "ALL VIEWS IN SCHEMA {}",
2858                    display_comma_separated(schemas)
2859                )
2860            }
2861            GrantObjects::AllFunctionsInSchema { schemas } => {
2862                write!(
2863                    f,
2864                    "ALL FUNCTIONS IN SCHEMA {}",
2865                    display_comma_separated(schemas)
2866                )
2867            }
2868            GrantObjects::AllSecretsInSchema { schemas } => {
2869                write!(
2870                    f,
2871                    "ALL SECRETS IN SCHEMA {}",
2872                    display_comma_separated(schemas)
2873                )
2874            }
2875            GrantObjects::AllSubscriptionsInSchema { schemas } => {
2876                write!(
2877                    f,
2878                    "ALL SUBSCRIPTIONS IN SCHEMA {}",
2879                    display_comma_separated(schemas)
2880                )
2881            }
2882            GrantObjects::AllConnectionsInSchema { schemas } => {
2883                write!(
2884                    f,
2885                    "ALL CONNECTIONS IN SCHEMA {}",
2886                    display_comma_separated(schemas)
2887                )
2888            }
2889            GrantObjects::Databases(databases) => {
2890                write!(f, "DATABASE {}", display_comma_separated(databases))
2891            }
2892            GrantObjects::Sources(sources) => {
2893                write!(f, "SOURCE {}", display_comma_separated(sources))
2894            }
2895            GrantObjects::Mviews(mviews) => {
2896                write!(f, "MATERIALIZED VIEW {}", display_comma_separated(mviews))
2897            }
2898            GrantObjects::Sinks(sinks) => {
2899                write!(f, "SINK {}", display_comma_separated(sinks))
2900            }
2901            GrantObjects::Views(views) => {
2902                write!(f, "VIEW {}", display_comma_separated(views))
2903            }
2904            GrantObjects::Connections(connections) => {
2905                write!(f, "CONNECTION {}", display_comma_separated(connections))
2906            }
2907            GrantObjects::Subscriptions(subscriptions) => {
2908                write!(f, "SUBSCRIPTION {}", display_comma_separated(subscriptions))
2909            }
2910            GrantObjects::Functions(func_descs) => {
2911                write!(f, "FUNCTION {}", display_comma_separated(func_descs))
2912            }
2913            GrantObjects::Secrets(secrets) => {
2914                write!(f, "SECRET {}", display_comma_separated(secrets))
2915            }
2916        }
2917    }
2918}
2919
2920#[derive(Debug, Clone, PartialEq, Eq, Hash)]
2921pub enum PrivilegeObjectType {
2922    Tables,
2923    Sources,
2924    Sinks,
2925    Mviews,
2926    Views,
2927    Functions,
2928    Connections,
2929    Secrets,
2930    Subscriptions,
2931    Schemas,
2932}
2933
2934impl fmt::Display for PrivilegeObjectType {
2935    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2936        match self {
2937            PrivilegeObjectType::Tables => f.write_str("TABLES")?,
2938            PrivilegeObjectType::Sources => f.write_str("SOURCES")?,
2939            PrivilegeObjectType::Sinks => f.write_str("SINKS")?,
2940            PrivilegeObjectType::Mviews => f.write_str("MATERIALIZED VIEWS")?,
2941            PrivilegeObjectType::Views => f.write_str("VIEWS")?,
2942            PrivilegeObjectType::Functions => f.write_str("FUNCTIONS")?,
2943            PrivilegeObjectType::Connections => f.write_str("CONNECTIONS")?,
2944            PrivilegeObjectType::Secrets => f.write_str("SECRETS")?,
2945            PrivilegeObjectType::Subscriptions => f.write_str("SUBSCRIPTIONS")?,
2946            PrivilegeObjectType::Schemas => f.write_str("SCHEMAS")?,
2947        };
2948        Ok(())
2949    }
2950}
2951
2952#[derive(Debug, Clone, PartialEq, Eq, Hash)]
2953pub enum DefaultPrivilegeOperation {
2954    Grant {
2955        privileges: Privileges,
2956        object_type: PrivilegeObjectType,
2957        grantees: Vec<Ident>,
2958        with_grant_option: bool,
2959    },
2960    Revoke {
2961        privileges: Privileges,
2962        object_type: PrivilegeObjectType,
2963        grantees: Vec<Ident>,
2964        revoke_grant_option: bool,
2965        cascade: bool,
2966    },
2967}
2968
2969impl fmt::Display for DefaultPrivilegeOperation {
2970    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2971        match self {
2972            DefaultPrivilegeOperation::Grant {
2973                privileges,
2974                object_type,
2975                grantees,
2976                with_grant_option,
2977            } => {
2978                write!(
2979                    f,
2980                    "GRANT {} ON {} TO {}",
2981                    privileges,
2982                    object_type,
2983                    display_comma_separated(grantees)
2984                )?;
2985                if *with_grant_option {
2986                    write!(f, " WITH GRANT OPTION")?;
2987                }
2988            }
2989            DefaultPrivilegeOperation::Revoke {
2990                privileges,
2991                object_type,
2992                grantees,
2993                revoke_grant_option,
2994                cascade,
2995            } => {
2996                write!(f, "REVOKE")?;
2997                if *revoke_grant_option {
2998                    write!(f, " GRANT OPTION FOR")?;
2999                }
3000                write!(
3001                    f,
3002                    " {} ON {} FROM {}",
3003                    privileges,
3004                    object_type,
3005                    display_comma_separated(grantees)
3006                )?;
3007                write!(f, " {}", if *cascade { "CASCADE" } else { "RESTRICT" })?;
3008            }
3009        }
3010        Ok(())
3011    }
3012}
3013
3014impl DefaultPrivilegeOperation {
3015    pub fn for_schemas(&self) -> bool {
3016        match &self {
3017            DefaultPrivilegeOperation::Grant { object_type, .. } => {
3018                object_type == &PrivilegeObjectType::Schemas
3019            }
3020            DefaultPrivilegeOperation::Revoke { object_type, .. } => {
3021                object_type == &PrivilegeObjectType::Schemas
3022            }
3023        }
3024    }
3025}
3026
3027#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3028pub enum AssignmentValue {
3029    /// An expression, e.g. `foo = 1`
3030    Expr(Expr),
3031    /// The `DEFAULT` keyword, e.g. `foo = DEFAULT`
3032    Default,
3033}
3034
3035impl fmt::Display for AssignmentValue {
3036    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3037        match self {
3038            AssignmentValue::Expr(expr) => write!(f, "{}", expr),
3039            AssignmentValue::Default => f.write_str("DEFAULT"),
3040        }
3041    }
3042}
3043
3044/// SQL assignment `foo = { expr | DEFAULT }` as used in SQLUpdate
3045#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3046pub struct Assignment {
3047    pub id: Vec<Ident>,
3048    pub value: AssignmentValue,
3049}
3050
3051impl fmt::Display for Assignment {
3052    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3053        write!(f, "{} = {}", display_separated(&self.id, "."), self.value)
3054    }
3055}
3056
3057#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3058pub enum FunctionArgExpr {
3059    Expr(Expr),
3060    /// Expr is an arbitrary expression, returning either a table or a column.
3061    /// Idents are the prefix of `*`, which are consecutive field accesses.
3062    /// e.g. `(table.v1).*` or `(table).v1.*`
3063    ExprQualifiedWildcard(Expr, Vec<Ident>),
3064    /// Qualified wildcard, e.g. `alias.*` or `schema.table.*`, followed by optional
3065    /// except syntax
3066    QualifiedWildcard(ObjectName, Option<Vec<Expr>>),
3067    /// An unqualified `*` or `* except (columns)`
3068    Wildcard(Option<Vec<Expr>>),
3069    /// A secret reference, e.g. `SECRET my_secret` or `SECRET my_secret AS FILE`
3070    SecretRef(SecretRefValue),
3071}
3072
3073impl fmt::Display for FunctionArgExpr {
3074    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3075        match self {
3076            FunctionArgExpr::Expr(expr) => write!(f, "{}", expr),
3077            FunctionArgExpr::ExprQualifiedWildcard(expr, prefix) => {
3078                write!(
3079                    f,
3080                    "({}){}.*",
3081                    expr,
3082                    prefix
3083                        .iter()
3084                        .format_with("", |i, f| f(&format_args!(".{i}")))
3085                )
3086            }
3087            FunctionArgExpr::QualifiedWildcard(prefix, except) => match except {
3088                Some(exprs) => write!(
3089                    f,
3090                    "{}.* EXCEPT ({})",
3091                    prefix,
3092                    exprs
3093                        .iter()
3094                        .map(|v| v.to_string())
3095                        .collect::<Vec<String>>()
3096                        .as_slice()
3097                        .join(", ")
3098                ),
3099                None => write!(f, "{}.*", prefix),
3100            },
3101
3102            FunctionArgExpr::SecretRef(secret_ref) => write!(f, "SECRET {}", secret_ref),
3103            FunctionArgExpr::Wildcard(except) => match except {
3104                Some(exprs) => write!(
3105                    f,
3106                    "* EXCEPT ({})",
3107                    exprs
3108                        .iter()
3109                        .map(|v| v.to_string())
3110                        .collect::<Vec<String>>()
3111                        .as_slice()
3112                        .join(", ")
3113                ),
3114                None => f.write_str("*"),
3115            },
3116        }
3117    }
3118}
3119
3120#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3121pub enum FunctionArg {
3122    Named { name: Ident, arg: FunctionArgExpr },
3123    Unnamed(FunctionArgExpr),
3124}
3125
3126impl FunctionArg {
3127    pub fn get_expr(&self) -> FunctionArgExpr {
3128        match self {
3129            FunctionArg::Named { name: _, arg } => arg.clone(),
3130            FunctionArg::Unnamed(arg) => arg.clone(),
3131        }
3132    }
3133}
3134
3135impl fmt::Display for FunctionArg {
3136    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3137        match self {
3138            FunctionArg::Named { name, arg } => write!(f, "{} => {}", name, arg),
3139            FunctionArg::Unnamed(unnamed_arg) => write!(f, "{}", unnamed_arg),
3140        }
3141    }
3142}
3143
3144/// A list of function arguments, including additional modifiers like `DISTINCT` or `ORDER BY`.
3145/// This basically holds all the information between the `(` and `)` in a function call.
3146#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3147pub struct FunctionArgList {
3148    /// Aggregate function calls may have a `DISTINCT`, e.g. `count(DISTINCT x)`.
3149    pub distinct: bool,
3150    pub args: Vec<FunctionArg>,
3151    /// Whether the last argument is variadic, e.g. `foo(a, b, VARIADIC c)`.
3152    pub variadic: bool,
3153    /// Aggregate function calls may have an `ORDER BY`, e.g. `array_agg(x ORDER BY y)`.
3154    pub order_by: Vec<OrderByExpr>,
3155    /// Window function calls may have an `IGNORE NULLS`, e.g. `first_value(x IGNORE NULLS)`.
3156    pub ignore_nulls: bool,
3157}
3158
3159impl fmt::Display for FunctionArgList {
3160    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3161        write!(f, "(")?;
3162        if self.distinct {
3163            write!(f, "DISTINCT ")?;
3164        }
3165        if self.variadic {
3166            for arg in &self.args[0..self.args.len() - 1] {
3167                write!(f, "{}, ", arg)?;
3168            }
3169            write!(f, "VARIADIC {}", self.args.last().unwrap())?;
3170        } else {
3171            write!(f, "{}", display_comma_separated(&self.args))?;
3172        }
3173        if !self.order_by.is_empty() {
3174            write!(f, " ORDER BY {}", display_comma_separated(&self.order_by))?;
3175        }
3176        if self.ignore_nulls {
3177            write!(f, " IGNORE NULLS")?;
3178        }
3179        write!(f, ")")?;
3180        Ok(())
3181    }
3182}
3183
3184impl FunctionArgList {
3185    pub fn empty() -> Self {
3186        Self {
3187            distinct: false,
3188            args: vec![],
3189            variadic: false,
3190            order_by: vec![],
3191            ignore_nulls: false,
3192        }
3193    }
3194
3195    pub fn args_only(args: Vec<FunctionArg>) -> Self {
3196        Self {
3197            distinct: false,
3198            args,
3199            variadic: false,
3200            order_by: vec![],
3201            ignore_nulls: false,
3202        }
3203    }
3204
3205    pub fn is_args_only(&self) -> bool {
3206        !self.distinct && !self.variadic && self.order_by.is_empty() && !self.ignore_nulls
3207    }
3208
3209    pub fn for_agg(distinct: bool, args: Vec<FunctionArg>, order_by: Vec<OrderByExpr>) -> Self {
3210        Self {
3211            distinct,
3212            args,
3213            variadic: false,
3214            order_by,
3215            ignore_nulls: false,
3216        }
3217    }
3218}
3219
3220/// A function call
3221#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3222pub struct Function {
3223    /// Whether the function is prefixed with `AGGREGATE:`
3224    pub scalar_as_agg: bool,
3225    /// Function name.
3226    pub name: ObjectName,
3227    /// Argument list of the function call, i.e. things in `()`.
3228    pub arg_list: FunctionArgList,
3229    /// `WITHIN GROUP` clause of the function call, for ordered-set aggregate functions.
3230    /// FIXME(rc): why we only support one expression here?
3231    pub within_group: Option<Box<OrderByExpr>>,
3232    /// `FILTER` clause of the function call, for aggregate and window (not supported yet) functions.
3233    pub filter: Option<Box<Expr>>,
3234    /// `OVER` clause of the function call, for window functions.
3235    pub over: Option<Window>,
3236}
3237
3238impl Function {
3239    pub fn no_arg(name: ObjectName) -> Self {
3240        Self {
3241            scalar_as_agg: false,
3242            name,
3243            arg_list: FunctionArgList::empty(),
3244            within_group: None,
3245            filter: None,
3246            over: None,
3247        }
3248    }
3249}
3250
3251impl fmt::Display for Function {
3252    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3253        if self.scalar_as_agg {
3254            write!(f, "AGGREGATE:")?;
3255        }
3256        write!(f, "{}{}", self.name, self.arg_list)?;
3257        if let Some(within_group) = &self.within_group {
3258            write!(f, " WITHIN GROUP (ORDER BY {})", within_group)?;
3259        }
3260        if let Some(filter) = &self.filter {
3261            write!(f, " FILTER (WHERE {})", filter)?;
3262        }
3263        if let Some(o) = &self.over {
3264            write!(f, " OVER {}", o)?;
3265        }
3266        Ok(())
3267    }
3268}
3269
3270#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3271pub enum ObjectType {
3272    Table,
3273    View,
3274    MaterializedView,
3275    Index,
3276    Schema,
3277    Source,
3278    Sink,
3279    Database,
3280    User,
3281    Connection,
3282    Secret,
3283    Subscription,
3284}
3285
3286impl fmt::Display for ObjectType {
3287    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3288        f.write_str(match self {
3289            ObjectType::Table => "TABLE",
3290            ObjectType::View => "VIEW",
3291            ObjectType::MaterializedView => "MATERIALIZED VIEW",
3292            ObjectType::Index => "INDEX",
3293            ObjectType::Schema => "SCHEMA",
3294            ObjectType::Source => "SOURCE",
3295            ObjectType::Sink => "SINK",
3296            ObjectType::Database => "DATABASE",
3297            ObjectType::User => "USER",
3298            ObjectType::Secret => "SECRET",
3299            ObjectType::Connection => "CONNECTION",
3300            ObjectType::Subscription => "SUBSCRIPTION",
3301        })
3302    }
3303}
3304
3305impl ParseTo for ObjectType {
3306    fn parse_to(parser: &mut Parser<'_>) -> ModalResult<Self> {
3307        let object_type = if parser.parse_keyword(Keyword::TABLE) {
3308            ObjectType::Table
3309        } else if parser.parse_keyword(Keyword::VIEW) {
3310            ObjectType::View
3311        } else if parser.parse_keywords(&[Keyword::MATERIALIZED, Keyword::VIEW]) {
3312            ObjectType::MaterializedView
3313        } else if parser.parse_keyword(Keyword::SOURCE) {
3314            ObjectType::Source
3315        } else if parser.parse_keyword(Keyword::SINK) {
3316            ObjectType::Sink
3317        } else if parser.parse_keyword(Keyword::INDEX) {
3318            ObjectType::Index
3319        } else if parser.parse_keyword(Keyword::SCHEMA) {
3320            ObjectType::Schema
3321        } else if parser.parse_keyword(Keyword::DATABASE) {
3322            ObjectType::Database
3323        } else if parser.parse_keyword(Keyword::USER) {
3324            ObjectType::User
3325        } else if parser.parse_keyword(Keyword::CONNECTION) {
3326            ObjectType::Connection
3327        } else if parser.parse_keyword(Keyword::SECRET) {
3328            ObjectType::Secret
3329        } else if parser.parse_keyword(Keyword::SUBSCRIPTION) {
3330            ObjectType::Subscription
3331        } else {
3332            return parser.expected(
3333                "TABLE, VIEW, INDEX, MATERIALIZED VIEW, SOURCE, SINK, SUBSCRIPTION, SCHEMA, DATABASE, USER, SECRET or CONNECTION after DROP",
3334            );
3335        };
3336        Ok(object_type)
3337    }
3338}
3339
3340#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3341pub struct SqlOption {
3342    pub name: ObjectName,
3343    pub value: SqlOptionValue,
3344}
3345
3346impl SqlOption {
3347    /// Creates an option whose value is a typed secret reference.
3348    pub fn from_secret_ref(name: &str, secret_ref: SecretRefValue) -> Self {
3349        Self {
3350            name: ObjectName(name.split('.').map(Ident::from_real_value).collect()),
3351            value: SqlOptionValue::SecretRef(secret_ref),
3352        }
3353    }
3354}
3355
3356impl fmt::Display for SqlOption {
3357    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3358        let should_redact = SQL_OPTION_REDACTION
3359            .try_with(|redaction| {
3360                let sql_option_name = self.name.real_value().to_lowercase();
3361                !redaction.allowlist.contains(&sql_option_name)
3362                    && redaction
3363                        .keywords
3364                        .iter()
3365                        .any(|k| sql_option_name.contains(k))
3366            })
3367            .unwrap_or(false);
3368        if should_redact {
3369            write!(f, "{} = [REDACTED]", self.name)
3370        } else {
3371            write!(f, "{} = {}", self.name, self.value)
3372        }
3373    }
3374}
3375
3376impl TryFrom<(&String, &String)> for SqlOption {
3377    type Error = ParserError;
3378
3379    fn try_from((name, value): (&String, &String)) -> Result<Self, Self::Error> {
3380        // Use from_real_value to properly escape the name, which handles cases like
3381        // "debezium.column.truncate.to.2000000.chars" where "2000000" would be
3382        // tokenized as a number instead of an identifier if not properly quoted.
3383        let name_parts: Vec<&str> = name.split('.').collect();
3384        let object_name = ObjectName(name_parts.into_iter().map(Ident::from_real_value).collect());
3385
3386        // Wrap the value in single quotes so it is always parsed as a string literal.
3387        // This prevents values containing special characters (e.g., "jdbc:postgresql://...")
3388        // from being incorrectly tokenized. Escape any embedded single quotes.
3389        let escaped_value = value.replace('\'', "''");
3390        let query = format!("{} = '{}'", object_name, escaped_value);
3391        let mut tokenizer = Tokenizer::new(query.as_str());
3392        let tokens = tokenizer.tokenize_with_location()?;
3393        let mut parser = Parser(&tokens);
3394        parser
3395            .parse_sql_option()
3396            .map_err(|e| ParserError::ParserError(e.to_string()))
3397    }
3398}
3399
3400#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3401pub enum SqlOptionValue {
3402    Value(Value),
3403    SecretRef(SecretRefValue),
3404    ConnectionRef(ConnectionRefValue),
3405    BackfillOrder(BackfillOrderStrategy),
3406}
3407
3408impl SqlOptionValue {
3409    /// Returns a `NULL` value.
3410    pub const fn null() -> Self {
3411        Self::Value(Value::Null)
3412    }
3413}
3414
3415impl fmt::Display for SqlOptionValue {
3416    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3417        match self {
3418            SqlOptionValue::Value(value) => write!(f, "{}", value),
3419            SqlOptionValue::SecretRef(secret_ref) => write!(f, "secret {}", secret_ref),
3420            SqlOptionValue::ConnectionRef(connection_ref) => {
3421                write!(f, "{}", connection_ref)
3422            }
3423            SqlOptionValue::BackfillOrder(order) => {
3424                write!(f, "{}", order)
3425            }
3426        }
3427    }
3428}
3429
3430impl From<Value> for SqlOptionValue {
3431    fn from(value: Value) -> Self {
3432        SqlOptionValue::Value(value)
3433    }
3434}
3435
3436#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3437pub enum EmitMode {
3438    Immediately,
3439    OnWindowClose,
3440}
3441
3442impl fmt::Display for EmitMode {
3443    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3444        f.write_str(match self {
3445            EmitMode::Immediately => "IMMEDIATELY",
3446            EmitMode::OnWindowClose => "ON WINDOW CLOSE",
3447        })
3448    }
3449}
3450
3451#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
3452pub enum OnConflict {
3453    UpdateFull,
3454    Nothing,
3455    UpdateIfNotNull,
3456}
3457
3458impl fmt::Display for OnConflict {
3459    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3460        f.write_str(match self {
3461            OnConflict::UpdateFull => "DO UPDATE FULL",
3462            OnConflict::Nothing => "DO NOTHING",
3463            OnConflict::UpdateIfNotNull => "DO UPDATE IF NOT NULL",
3464        })
3465    }
3466}
3467
3468#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3469pub enum Engine {
3470    Hummock,
3471    Iceberg,
3472}
3473
3474impl fmt::Display for crate::ast::Engine {
3475    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3476        f.write_str(match self {
3477            crate::ast::Engine::Hummock => "HUMMOCK",
3478            crate::ast::Engine::Iceberg => "ICEBERG",
3479        })
3480    }
3481}
3482
3483#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3484pub enum SetTimeZoneValue {
3485    Ident(Ident),
3486    Literal(Value),
3487    Local,
3488    Default,
3489}
3490
3491impl fmt::Display for SetTimeZoneValue {
3492    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3493        match self {
3494            SetTimeZoneValue::Ident(ident) => write!(f, "{}", ident),
3495            SetTimeZoneValue::Literal(value) => write!(f, "{}", value),
3496            SetTimeZoneValue::Local => f.write_str("LOCAL"),
3497            SetTimeZoneValue::Default => f.write_str("DEFAULT"),
3498        }
3499    }
3500}
3501
3502#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3503pub enum TransactionMode {
3504    AccessMode(TransactionAccessMode),
3505    IsolationLevel(TransactionIsolationLevel),
3506}
3507
3508impl fmt::Display for TransactionMode {
3509    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3510        use TransactionMode::*;
3511        match self {
3512            AccessMode(access_mode) => write!(f, "{}", access_mode),
3513            IsolationLevel(iso_level) => write!(f, "ISOLATION LEVEL {}", iso_level),
3514        }
3515    }
3516}
3517
3518#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3519pub enum TransactionAccessMode {
3520    ReadOnly,
3521    ReadWrite,
3522}
3523
3524impl fmt::Display for TransactionAccessMode {
3525    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3526        use TransactionAccessMode::*;
3527        f.write_str(match self {
3528            ReadOnly => "READ ONLY",
3529            ReadWrite => "READ WRITE",
3530        })
3531    }
3532}
3533
3534#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3535pub enum TransactionIsolationLevel {
3536    ReadUncommitted,
3537    ReadCommitted,
3538    RepeatableRead,
3539    Serializable,
3540}
3541
3542impl fmt::Display for TransactionIsolationLevel {
3543    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3544        use TransactionIsolationLevel::*;
3545        f.write_str(match self {
3546            ReadUncommitted => "READ UNCOMMITTED",
3547            ReadCommitted => "READ COMMITTED",
3548            RepeatableRead => "REPEATABLE READ",
3549            Serializable => "SERIALIZABLE",
3550        })
3551    }
3552}
3553
3554#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3555pub enum ShowStatementFilter {
3556    Like(String),
3557    ILike(String),
3558    Where(Expr),
3559}
3560
3561impl fmt::Display for ShowStatementFilter {
3562    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3563        use ShowStatementFilter::*;
3564        match self {
3565            Like(pattern) => write!(f, "LIKE '{}'", value::escape_single_quote_string(pattern)),
3566            ILike(pattern) => write!(f, "ILIKE {}", value::escape_single_quote_string(pattern)),
3567            Where(expr) => write!(f, "WHERE {}", expr),
3568        }
3569    }
3570}
3571
3572/// Function describe in DROP FUNCTION.
3573#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
3574pub enum DropFunctionOption {
3575    Restrict,
3576    Cascade,
3577}
3578
3579impl fmt::Display for DropFunctionOption {
3580    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3581        match self {
3582            DropFunctionOption::Restrict => write!(f, "RESTRICT "),
3583            DropFunctionOption::Cascade => write!(f, "CASCADE  "),
3584        }
3585    }
3586}
3587
3588/// Function describe in DROP FUNCTION.
3589#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3590pub struct FunctionDesc {
3591    pub name: ObjectName,
3592    pub args: Option<Vec<OperateFunctionArg>>,
3593}
3594
3595impl fmt::Display for FunctionDesc {
3596    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3597        write!(f, "{}", self.name)?;
3598        if let Some(args) = &self.args {
3599            write!(f, "({})", display_comma_separated(args))?;
3600        }
3601        Ok(())
3602    }
3603}
3604
3605/// Function argument in CREATE FUNCTION.
3606#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3607pub struct OperateFunctionArg {
3608    pub mode: Option<ArgMode>,
3609    pub name: Option<Ident>,
3610    pub data_type: DataType,
3611    pub default_expr: Option<Expr>,
3612}
3613
3614impl OperateFunctionArg {
3615    /// Returns an unnamed argument.
3616    pub fn unnamed(data_type: DataType) -> Self {
3617        Self {
3618            mode: None,
3619            name: None,
3620            data_type,
3621            default_expr: None,
3622        }
3623    }
3624
3625    /// Returns an argument with name.
3626    pub fn with_name(name: &str, data_type: DataType) -> Self {
3627        Self {
3628            mode: None,
3629            name: Some(name.into()),
3630            data_type,
3631            default_expr: None,
3632        }
3633    }
3634}
3635
3636impl fmt::Display for OperateFunctionArg {
3637    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3638        if let Some(mode) = &self.mode {
3639            write!(f, "{} ", mode)?;
3640        }
3641        if let Some(name) = &self.name {
3642            write!(f, "{} ", name)?;
3643        }
3644        write!(f, "{}", self.data_type)?;
3645        if let Some(default_expr) = &self.default_expr {
3646            write!(f, " = {}", default_expr)?;
3647        }
3648        Ok(())
3649    }
3650}
3651
3652/// The mode of an argument in CREATE FUNCTION.
3653#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3654pub enum ArgMode {
3655    In,
3656    Out,
3657    InOut,
3658}
3659
3660impl fmt::Display for ArgMode {
3661    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3662        match self {
3663            ArgMode::In => write!(f, "IN"),
3664            ArgMode::Out => write!(f, "OUT"),
3665            ArgMode::InOut => write!(f, "INOUT"),
3666        }
3667    }
3668}
3669
3670/// These attributes inform the query optimizer about the behavior of the function.
3671#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3672pub enum FunctionBehavior {
3673    Immutable,
3674    Stable,
3675    Volatile,
3676}
3677
3678impl fmt::Display for FunctionBehavior {
3679    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3680        match self {
3681            FunctionBehavior::Immutable => write!(f, "IMMUTABLE"),
3682            FunctionBehavior::Stable => write!(f, "STABLE"),
3683            FunctionBehavior::Volatile => write!(f, "VOLATILE"),
3684        }
3685    }
3686}
3687
3688#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
3689pub enum FunctionDefinition {
3690    Identifier(String),
3691    SingleQuotedDef(String),
3692    DoubleDollarDef(String),
3693}
3694
3695impl fmt::Display for FunctionDefinition {
3696    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3697        match self {
3698            FunctionDefinition::Identifier(s) => write!(f, "{s}")?,
3699            FunctionDefinition::SingleQuotedDef(s) => write!(f, "'{s}'")?,
3700            FunctionDefinition::DoubleDollarDef(s) => write!(f, "$${s}$$")?,
3701        }
3702        Ok(())
3703    }
3704}
3705
3706impl FunctionDefinition {
3707    /// Returns the function definition as a string slice.
3708    pub fn as_str(&self) -> &str {
3709        match self {
3710            FunctionDefinition::Identifier(s) => s,
3711            FunctionDefinition::SingleQuotedDef(s) => s,
3712            FunctionDefinition::DoubleDollarDef(s) => s,
3713        }
3714    }
3715
3716    /// Returns the function definition as a string.
3717    pub fn into_string(self) -> String {
3718        match self {
3719            FunctionDefinition::Identifier(s) => s,
3720            FunctionDefinition::SingleQuotedDef(s) => s,
3721            FunctionDefinition::DoubleDollarDef(s) => s,
3722        }
3723    }
3724}
3725
3726/// Return types of a function.
3727#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3728pub enum CreateFunctionReturns {
3729    /// RETURNS rettype
3730    Value(DataType),
3731    /// RETURNS TABLE ( column_name column_type [, ...] )
3732    Table(Vec<TableColumnDef>),
3733}
3734
3735impl fmt::Display for CreateFunctionReturns {
3736    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3737        match self {
3738            Self::Value(data_type) => write!(f, "RETURNS {}", data_type),
3739            Self::Table(columns) => {
3740                write!(f, "RETURNS TABLE ({})", display_comma_separated(columns))
3741            }
3742        }
3743    }
3744}
3745
3746/// Table column definition
3747#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3748pub struct TableColumnDef {
3749    pub name: Ident,
3750    pub data_type: DataType,
3751}
3752
3753impl fmt::Display for TableColumnDef {
3754    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3755        write!(f, "{} {}", self.name, self.data_type)
3756    }
3757}
3758
3759/// Postgres specific feature.
3760///
3761/// See [Postgresdocs](https://www.postgresql.org/docs/15/sql-createfunction.html)
3762/// for more details
3763#[derive(Debug, Default, Clone, PartialEq, Eq, Hash)]
3764pub struct CreateFunctionBody {
3765    /// LANGUAGE lang_name
3766    pub language: Option<Ident>,
3767    /// RUNTIME runtime_name
3768    pub runtime: Option<Ident>,
3769
3770    /// IMMUTABLE | STABLE | VOLATILE
3771    pub behavior: Option<FunctionBehavior>,
3772    /// AS 'definition'
3773    ///
3774    /// Note that Hive's `AS class_name` is also parsed here.
3775    pub as_: Option<FunctionDefinition>,
3776    /// RETURN expression
3777    pub return_: Option<Expr>,
3778    /// USING ...
3779    pub using: Option<CreateFunctionUsing>,
3780}
3781
3782impl fmt::Display for CreateFunctionBody {
3783    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3784        if let Some(language) = &self.language {
3785            write!(f, " LANGUAGE {language}")?;
3786        }
3787        if let Some(runtime) = &self.runtime {
3788            write!(f, " RUNTIME {runtime}")?;
3789        }
3790        if let Some(behavior) = &self.behavior {
3791            write!(f, " {behavior}")?;
3792        }
3793        if let Some(definition) = &self.as_ {
3794            write!(f, " AS {definition}")?;
3795        }
3796        if let Some(expr) = &self.return_ {
3797            write!(f, " RETURN {expr}")?;
3798        }
3799        if let Some(using) = &self.using {
3800            write!(f, " {using}")?;
3801        }
3802        Ok(())
3803    }
3804}
3805
3806#[derive(Debug, Default, Clone, PartialEq, Eq, Hash)]
3807pub struct CreateFunctionWithOptions {
3808    /// Always retry on network errors.
3809    pub always_retry_on_network_error: Option<bool>,
3810    /// Use async functions (only available for JS UDF)
3811    pub r#async: Option<bool>,
3812    /// Call in batch mode (only available for JS UDF)
3813    pub batch: Option<bool>,
3814}
3815
3816/// TODO(kwannoel): Generate from the struct definition instead.
3817impl TryFrom<Vec<SqlOption>> for CreateFunctionWithOptions {
3818    type Error = StrError;
3819
3820    fn try_from(with_options: Vec<SqlOption>) -> Result<Self, Self::Error> {
3821        let mut options = Self::default();
3822        for option in with_options {
3823            match option.name.to_string().to_lowercase().as_str() {
3824                "always_retry_on_network_error" => {
3825                    options.always_retry_on_network_error = Some(matches!(
3826                        option.value,
3827                        SqlOptionValue::Value(Value::Boolean(true))
3828                    ));
3829                }
3830                "async" => {
3831                    options.r#async = Some(matches!(
3832                        option.value,
3833                        SqlOptionValue::Value(Value::Boolean(true))
3834                    ))
3835                }
3836                "batch" => {
3837                    options.batch = Some(matches!(
3838                        option.value,
3839                        SqlOptionValue::Value(Value::Boolean(true))
3840                    ))
3841                }
3842                _ => {
3843                    return Err(StrError(format!("unknown option: {}", option.name)));
3844                }
3845            }
3846        }
3847        Ok(options)
3848    }
3849}
3850
3851impl Display for CreateFunctionWithOptions {
3852    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3853        if self == &Self::default() {
3854            return Ok(());
3855        }
3856        let mut options = vec![];
3857        if let Some(v) = self.always_retry_on_network_error {
3858            options.push(format!("always_retry_on_network_error = {}", v));
3859        }
3860        if let Some(v) = self.r#async {
3861            options.push(format!("async = {}", v));
3862        }
3863        if let Some(v) = self.batch {
3864            options.push(format!("batch = {}", v));
3865        }
3866        write!(f, " WITH ( {} )", display_comma_separated(&options))
3867    }
3868}
3869
3870#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
3871pub enum CreateFunctionUsing {
3872    Link(String),
3873    Base64(String),
3874}
3875
3876impl fmt::Display for CreateFunctionUsing {
3877    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3878        write!(f, "USING ")?;
3879        match self {
3880            CreateFunctionUsing::Link(uri) => write!(f, "LINK '{uri}'"),
3881            CreateFunctionUsing::Base64(s) => {
3882                write!(f, "BASE64 '{s}'")
3883            }
3884        }
3885    }
3886}
3887
3888#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3889pub struct ConfigParam {
3890    pub param: Ident,
3891    pub value: SetVariableValue,
3892}
3893
3894impl fmt::Display for ConfigParam {
3895    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3896        write!(f, "SET {} = {}", self.param, self.value)
3897    }
3898}
3899
3900#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3901pub enum SetVariableValue {
3902    Single(SetVariableValueSingle),
3903    List(Vec<SetVariableValueSingle>),
3904    Default,
3905}
3906
3907impl From<SetVariableValueSingle> for SetVariableValue {
3908    fn from(value: SetVariableValueSingle) -> Self {
3909        SetVariableValue::Single(value)
3910    }
3911}
3912
3913impl fmt::Display for SetVariableValue {
3914    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3915        use SetVariableValue::*;
3916        match self {
3917            Single(val) => write!(f, "{}", val),
3918            List(list) => write!(f, "{}", display_comma_separated(list),),
3919            Default => write!(f, "DEFAULT"),
3920        }
3921    }
3922}
3923
3924#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3925pub enum SetVariableValueSingle {
3926    Ident(Ident),
3927    Literal(Value),
3928    Raw(String),
3929}
3930
3931impl SetVariableValueSingle {
3932    pub fn to_string_unquoted(&self) -> String {
3933        match self {
3934            Self::Literal(Value::SingleQuotedString(s))
3935            | Self::Literal(Value::DoubleQuotedString(s))
3936            | Self::Raw(s) => s.clone(),
3937            _ => self.to_string(),
3938        }
3939    }
3940}
3941
3942impl fmt::Display for SetVariableValueSingle {
3943    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3944        use SetVariableValueSingle::*;
3945        match self {
3946            Ident(ident) => write!(f, "{}", ident),
3947            Literal(literal) => write!(f, "{}", literal),
3948            Raw(raw) => write!(f, "{}", raw),
3949        }
3950    }
3951}
3952
3953#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3954pub enum AsOf {
3955    ProcessTime,
3956    /// Internal marker for a process-time temporal join whose lookup side is broadcast to all join
3957    /// actors. It stays on the lookup relation so optimizer rewrites cannot detach the strategy
3958    /// from that relation; [`crate::ast::Join`]'s `Display` renders the modifier in join position.
3959    ProcessTimeBroadcast,
3960    // used by time travel
3961    ProcessTimeWithInterval((String, DateTimeField)),
3962    // the number of seconds that have elapsed since the Unix epoch, which is January 1, 1970 at 00:00:00 Coordinated Universal Time (UTC).
3963    TimestampNum(i64),
3964    TimestampString(String),
3965    VersionNum(i64),
3966    VersionString(String),
3967}
3968
3969impl fmt::Display for AsOf {
3970    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3971        use AsOf::*;
3972        match self {
3973            ProcessTime | ProcessTimeBroadcast => {
3974                write!(f, " FOR SYSTEM_TIME AS OF PROCTIME()")
3975            }
3976            ProcessTimeWithInterval((value, leading_field)) => write!(
3977                f,
3978                " FOR SYSTEM_TIME AS OF NOW() - '{}' {}",
3979                value, leading_field
3980            ),
3981            TimestampNum(ts) => write!(f, " FOR SYSTEM_TIME AS OF {}", ts),
3982            TimestampString(ts) => write!(f, " FOR SYSTEM_TIME AS OF '{}'", ts),
3983            VersionNum(v) => write!(f, " FOR SYSTEM_VERSION AS OF {}", v),
3984            VersionString(v) => write!(f, " FOR SYSTEM_VERSION AS OF '{}'", v),
3985        }
3986    }
3987}
3988
3989#[derive(Debug, Clone, PartialEq, Eq, Hash)]
3990pub enum DiscardType {
3991    All,
3992}
3993
3994impl fmt::Display for DiscardType {
3995    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3996        use DiscardType::*;
3997        match self {
3998            All => write!(f, "ALL"),
3999        }
4000    }
4001}
4002
4003// We decouple "default" from none,
4004// so we can choose strategies that make the most sense.
4005#[derive(Debug, Default, Clone, PartialEq, Eq, Hash)]
4006pub enum BackfillOrderStrategy {
4007    #[default]
4008    Default,
4009    None,
4010    Auto,
4011    Fixed(Vec<(ObjectName, ObjectName)>),
4012}
4013
4014impl fmt::Display for BackfillOrderStrategy {
4015    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
4016        use BackfillOrderStrategy::*;
4017        match self {
4018            Default => write!(f, "DEFAULT"),
4019            None => write!(f, "NONE"),
4020            Auto => write!(f, "AUTO"),
4021            Fixed(map) => {
4022                let mut parts = vec![];
4023                for (start, end) in map {
4024                    parts.push(format!("{} -> {}", start, end));
4025                }
4026                write!(f, "FIXED({})", display_comma_separated(&parts))
4027            }
4028        }
4029    }
4030}
4031
4032impl Statement {
4033    pub fn to_redacted_string(&self, redaction: SqlOptionRedactionRef) -> String {
4034        SQL_OPTION_REDACTION.sync_scope(redaction, || self.to_string_unchecked())
4035    }
4036
4037    /// Create a new `CREATE TABLE` statement with the given `name` and empty fields.
4038    pub fn default_create_table(name: ObjectName) -> Self {
4039        Self::CreateTable {
4040            name,
4041            or_replace: false,
4042            temporary: false,
4043            if_not_exists: false,
4044            columns: Vec::new(),
4045            wildcard_idx: None,
4046            constraints: Vec::new(),
4047            with_options: Vec::new(),
4048            format_encode: None,
4049            source_watermarks: Vec::new(),
4050            append_only: false,
4051            on_conflict: None,
4052            with_version_columns: Vec::new(),
4053            query: None,
4054            cdc_table_info: None,
4055            include_column_options: Vec::new(),
4056            webhook_info: None,
4057            engine: Engine::Hummock,
4058        }
4059    }
4060}
4061
4062#[cfg(test)]
4063mod tests {
4064    use super::*;
4065
4066    #[test]
4067    fn test_sql_option_redaction_preserves_primary_key() {
4068        let redaction = Arc::new(SqlOptionRedaction::new(
4069            [
4070                "credential".into(),
4071                "key".into(),
4072                "password".into(),
4073                "private".into(),
4074                "secret".into(),
4075                "token".into(),
4076            ],
4077            ["primary_key".into()],
4078        ));
4079        for (name, expected_value) in [
4080            ("primary_key", "'id'"),
4081            ("\"primary_key\"", "'id'"),
4082            ("\"PRIMARY_KEY\"", "'id'"),
4083            ("api_key", "[REDACTED]"),
4084            ("access_key", "[REDACTED]"),
4085            ("password", "[REDACTED]"),
4086            ("token", "[REDACTED]"),
4087            ("primary_key_password", "[REDACTED]"),
4088            ("my_primary_key", "[REDACTED]"),
4089            ("aws.secret_access_key", "[REDACTED]"),
4090            ("connector", "'id'"),
4091        ] {
4092            let sql = format!("CREATE SINK s FROM t WITH ({name} = 'id')");
4093            let statement = Parser::parse_sql(&sql).unwrap().remove(0);
4094            assert_eq!(
4095                statement.to_redacted_string(redaction.clone()),
4096                format!("CREATE SINK s FROM t WITH ({name} = {expected_value})"),
4097                "option: {name}",
4098            );
4099            assert_eq!(statement.to_string_unchecked(), sql);
4100        }
4101    }
4102
4103    #[test]
4104    fn test_sql_option_redaction_custom_allowlist() {
4105        let sql = "CREATE SINK s FROM t WITH (primary_key = 'id', partition_key = 'region', partition_key_password = 'secret')";
4106        let statement = Parser::parse_sql(sql).unwrap().remove(0);
4107        for (allowlist, expected) in [
4108            (
4109                vec![],
4110                "CREATE SINK s FROM t WITH (primary_key = [REDACTED], partition_key = [REDACTED], partition_key_password = [REDACTED])",
4111            ),
4112            (
4113                vec!["PARTITION_KEY".into()],
4114                "CREATE SINK s FROM t WITH (primary_key = [REDACTED], partition_key = 'region', partition_key_password = [REDACTED])",
4115            ),
4116        ] {
4117            let redaction = Arc::new(SqlOptionRedaction::new(["KEY".into()], allowlist));
4118            assert_eq!(statement.to_redacted_string(redaction), expected);
4119        }
4120    }
4121
4122    #[test]
4123    fn test_grouping_sets_display() {
4124        // a and b in different group
4125        let grouping_sets = Expr::GroupingSets(vec![
4126            vec![Expr::Identifier(Ident::new_unchecked("a"))],
4127            vec![Expr::Identifier(Ident::new_unchecked("b"))],
4128        ]);
4129        assert_eq!("GROUPING SETS ((a), (b))", format!("{}", grouping_sets));
4130
4131        // a and b in the same group
4132        let grouping_sets = Expr::GroupingSets(vec![vec![
4133            Expr::Identifier(Ident::new_unchecked("a")),
4134            Expr::Identifier(Ident::new_unchecked("b")),
4135        ]]);
4136        assert_eq!("GROUPING SETS ((a, b))", format!("{}", grouping_sets));
4137
4138        // (a, b) and (c, d) in different group
4139        let grouping_sets = Expr::GroupingSets(vec![
4140            vec![
4141                Expr::Identifier(Ident::new_unchecked("a")),
4142                Expr::Identifier(Ident::new_unchecked("b")),
4143            ],
4144            vec![
4145                Expr::Identifier(Ident::new_unchecked("c")),
4146                Expr::Identifier(Ident::new_unchecked("d")),
4147            ],
4148        ]);
4149        assert_eq!(
4150            "GROUPING SETS ((a, b), (c, d))",
4151            format!("{}", grouping_sets)
4152        );
4153    }
4154
4155    #[test]
4156    fn test_rollup_display() {
4157        let rollup = Expr::Rollup(vec![vec![Expr::Identifier(Ident::new_unchecked("a"))]]);
4158        assert_eq!("ROLLUP (a)", format!("{}", rollup));
4159
4160        let rollup = Expr::Rollup(vec![vec![
4161            Expr::Identifier(Ident::new_unchecked("a")),
4162            Expr::Identifier(Ident::new_unchecked("b")),
4163        ]]);
4164        assert_eq!("ROLLUP ((a, b))", format!("{}", rollup));
4165
4166        let rollup = Expr::Rollup(vec![
4167            vec![Expr::Identifier(Ident::new_unchecked("a"))],
4168            vec![Expr::Identifier(Ident::new_unchecked("b"))],
4169        ]);
4170        assert_eq!("ROLLUP (a, b)", format!("{}", rollup));
4171
4172        let rollup = Expr::Rollup(vec![
4173            vec![Expr::Identifier(Ident::new_unchecked("a"))],
4174            vec![
4175                Expr::Identifier(Ident::new_unchecked("b")),
4176                Expr::Identifier(Ident::new_unchecked("c")),
4177            ],
4178            vec![Expr::Identifier(Ident::new_unchecked("d"))],
4179        ]);
4180        assert_eq!("ROLLUP (a, (b, c), d)", format!("{}", rollup));
4181    }
4182
4183    #[test]
4184    fn test_cube_display() {
4185        let cube = Expr::Cube(vec![vec![Expr::Identifier(Ident::new_unchecked("a"))]]);
4186        assert_eq!("CUBE (a)", format!("{}", cube));
4187
4188        let cube = Expr::Cube(vec![vec![
4189            Expr::Identifier(Ident::new_unchecked("a")),
4190            Expr::Identifier(Ident::new_unchecked("b")),
4191        ]]);
4192        assert_eq!("CUBE ((a, b))", format!("{}", cube));
4193
4194        let cube = Expr::Cube(vec![
4195            vec![Expr::Identifier(Ident::new_unchecked("a"))],
4196            vec![Expr::Identifier(Ident::new_unchecked("b"))],
4197        ]);
4198        assert_eq!("CUBE (a, b)", format!("{}", cube));
4199
4200        let cube = Expr::Cube(vec![
4201            vec![Expr::Identifier(Ident::new_unchecked("a"))],
4202            vec![
4203                Expr::Identifier(Ident::new_unchecked("b")),
4204                Expr::Identifier(Ident::new_unchecked("c")),
4205            ],
4206            vec![Expr::Identifier(Ident::new_unchecked("d"))],
4207        ]);
4208        assert_eq!("CUBE (a, (b, c), d)", format!("{}", cube));
4209    }
4210
4211    #[test]
4212    fn test_array_index_display() {
4213        let array_index = Expr::Index {
4214            obj: Box::new(Expr::Identifier(Ident::new_unchecked("v1"))),
4215            index: Box::new(Expr::Value(Value::Number("1".into()))),
4216        };
4217        assert_eq!("v1[1]", format!("{}", array_index));
4218
4219        let array_index2 = Expr::Index {
4220            obj: Box::new(array_index),
4221            index: Box::new(Expr::Value(Value::Number("1".into()))),
4222        };
4223        assert_eq!("v1[1][1]", format!("{}", array_index2));
4224    }
4225
4226    #[test]
4227    /// issue: https://github.com/risingwavelabs/risingwave/issues/7635
4228    fn test_nested_op_display() {
4229        let binary_op = Expr::BinaryOp {
4230            left: Box::new(Expr::Value(Value::Boolean(true))),
4231            op: BinaryOperator::Or,
4232            right: Box::new(Expr::IsNotFalse(Box::new(Expr::Value(Value::Boolean(
4233                true,
4234            ))))),
4235        };
4236        assert_eq!("true OR true IS NOT FALSE", format!("{}", binary_op));
4237
4238        let unary_op = Expr::UnaryOp {
4239            op: UnaryOperator::Not,
4240            expr: Box::new(Expr::IsNotFalse(Box::new(Expr::Value(Value::Boolean(
4241                true,
4242            ))))),
4243        };
4244        assert_eq!("NOT true IS NOT FALSE", format!("{}", unary_op));
4245    }
4246
4247    #[test]
4248    fn test_create_function_display() {
4249        let create_function = Statement::CreateFunction {
4250            or_replace: false,
4251            temporary: false,
4252            if_not_exists: false,
4253            name: ObjectName(vec![Ident::new_unchecked("foo")]),
4254            args: Some(vec![OperateFunctionArg::unnamed(DataType::Int)]),
4255            returns: Some(CreateFunctionReturns::Value(DataType::Int)),
4256            params: CreateFunctionBody {
4257                language: Some(Ident::new_unchecked("python")),
4258                runtime: None,
4259                behavior: Some(FunctionBehavior::Immutable),
4260                as_: Some(FunctionDefinition::SingleQuotedDef("SELECT 1".to_owned())),
4261                return_: None,
4262                using: None,
4263            },
4264            with_options: CreateFunctionWithOptions {
4265                always_retry_on_network_error: None,
4266                r#async: None,
4267                batch: None,
4268            },
4269        };
4270        assert_eq!(
4271            "CREATE FUNCTION foo(INT) RETURNS INT LANGUAGE python IMMUTABLE AS 'SELECT 1'",
4272            format!("{}", create_function)
4273        );
4274        let create_function = Statement::CreateFunction {
4275            or_replace: false,
4276            temporary: false,
4277            if_not_exists: false,
4278            name: ObjectName(vec![Ident::new_unchecked("foo")]),
4279            args: Some(vec![OperateFunctionArg::unnamed(DataType::Int)]),
4280            returns: Some(CreateFunctionReturns::Value(DataType::Int)),
4281            params: CreateFunctionBody {
4282                language: Some(Ident::new_unchecked("python")),
4283                runtime: None,
4284                behavior: Some(FunctionBehavior::Immutable),
4285                as_: Some(FunctionDefinition::SingleQuotedDef("SELECT 1".to_owned())),
4286                return_: None,
4287                using: None,
4288            },
4289            with_options: CreateFunctionWithOptions {
4290                always_retry_on_network_error: Some(true),
4291                r#async: None,
4292                batch: None,
4293            },
4294        };
4295        assert_eq!(
4296            "CREATE FUNCTION foo(INT) RETURNS INT LANGUAGE python IMMUTABLE AS 'SELECT 1' WITH ( always_retry_on_network_error = true )",
4297            format!("{}", create_function)
4298        );
4299    }
4300
4301    #[test]
4302    fn test_sql_option_from_secret_ref() {
4303        let text_option = SqlOption::from_secret_ref(
4304            "password",
4305            SecretRefValue {
4306                secret_name: ObjectName(vec![
4307                    Ident::from_real_value("public"),
4308                    Ident::from_real_value("s2"),
4309                ]),
4310                ref_as: SecretRefAsType::Text,
4311            },
4312        );
4313
4314        assert!(matches!(text_option.value, SqlOptionValue::SecretRef(_)));
4315        assert_eq!(text_option.to_string(), "password = secret public.s2");
4316
4317        let file_option = SqlOption::from_secret_ref(
4318            "certificate",
4319            SecretRefValue {
4320                secret_name: ObjectName(vec![Ident::from_real_value("cert")]),
4321                ref_as: SecretRefAsType::File,
4322            },
4323        );
4324
4325        assert_eq!(file_option.to_string(), "certificate = secret cert AS FILE");
4326    }
4327}
risingwave_sqlparser/ast/mod.rs

risingwave_sqlparser/ast/
mod.rs
