mz_sql_parser/ast/defs/
ddl.rs

1// Copyright 2018 sqlparser-rs contributors. All rights reserved.
2// Copyright Materialize, Inc. and contributors. All rights reserved.
3//
4// This file is derived from the sqlparser-rs project, available at
5// https://github.com/andygrove/sqlparser-rs. It was incorporated
6// directly into Materialize on December 21, 2019.
7//
8// Licensed under the Apache License, Version 2.0 (the "License");
9// you may not use this file except in compliance with the License.
10// You may obtain a copy of the License in the LICENSE file at the
11// root of this repository, or online at
12//
13//     http://www.apache.org/licenses/LICENSE-2.0
14//
15// Unless required by applicable law or agreed to in writing, software
16// distributed under the License is distributed on an "AS IS" BASIS,
17// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
18// See the License for the specific language governing permissions and
19// limitations under the License.
20
21//! AST types specific to CREATE/ALTER variants of [crate::ast::Statement]
22//! (commonly referred to as Data Definition Language, or DDL)
23
24use std::fmt;
25
26use crate::ast::display::{self, AstDisplay, AstFormatter, WithOptionName};
27use crate::ast::{
28    AstInfo, ColumnName, Expr, Ident, OrderByExpr, UnresolvedItemName, Version, WithOptionValue,
29};
30
31#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
32pub enum MaterializedViewOptionName {
33    /// The `ASSERT NOT NULL [=] <ident>` option.
34    AssertNotNull,
35    PartitionBy,
36    RetainHistory,
37    /// The `REFRESH [=] ...` option.
38    Refresh,
39}
40
41impl AstDisplay for MaterializedViewOptionName {
42    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
43        match self {
44            MaterializedViewOptionName::AssertNotNull => f.write_str("ASSERT NOT NULL"),
45            MaterializedViewOptionName::PartitionBy => f.write_str("PARTITION BY"),
46            MaterializedViewOptionName::RetainHistory => f.write_str("RETAIN HISTORY"),
47            MaterializedViewOptionName::Refresh => f.write_str("REFRESH"),
48        }
49    }
50}
51
52impl WithOptionName for MaterializedViewOptionName {
53    /// # WARNING
54    ///
55    /// Whenever implementing this trait consider very carefully whether or not
56    /// this value could contain sensitive user data. If you're uncertain, err
57    /// on the conservative side and return `true`.
58    fn redact_value(&self) -> bool {
59        match self {
60            MaterializedViewOptionName::AssertNotNull
61            | MaterializedViewOptionName::RetainHistory
62            | MaterializedViewOptionName::Refresh => false,
63            // The value is an arbitrary user expression/literal that may embed
64            // sensitive data, so redact it (mirrors `KafkaSinkConfigOptionName`).
65            MaterializedViewOptionName::PartitionBy => true,
66        }
67    }
68}
69
70#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
71pub struct MaterializedViewOption<T: AstInfo> {
72    pub name: MaterializedViewOptionName,
73    pub value: Option<WithOptionValue<T>>,
74}
75impl_display_for_with_option!(MaterializedViewOption);
76
77#[derive(Debug, Clone, PartialEq, Eq, Hash)]
78pub struct Schema {
79    pub schema: String,
80}
81
82impl AstDisplay for Schema {
83    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
84        f.write_str("SCHEMA '");
85        f.write_node(&display::escape_single_quote_string(&self.schema));
86        f.write_str("'");
87    }
88}
89impl_display!(Schema);
90
91#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
92pub enum AvroSchemaOptionName {
93    /// The `CONFLUENT WIRE FORMAT [=] <bool>` option.
94    ConfluentWireFormat,
95}
96
97impl AstDisplay for AvroSchemaOptionName {
98    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
99        match self {
100            AvroSchemaOptionName::ConfluentWireFormat => f.write_str("CONFLUENT WIRE FORMAT"),
101        }
102    }
103}
104
105impl WithOptionName for AvroSchemaOptionName {
106    /// # WARNING
107    ///
108    /// Whenever implementing this trait consider very carefully whether or not
109    /// this value could contain sensitive user data. If you're uncertain, err
110    /// on the conservative side and return `true`.
111    fn redact_value(&self) -> bool {
112        match self {
113            Self::ConfluentWireFormat => false,
114        }
115    }
116}
117
118/// Options accepted on the `USING AWS GLUE SCHEMA REGISTRY CONNECTION <name> (…)`
119/// form. Today there is only `SCHEMA NAME`, which is required (the Glue
120/// purification step needs it to fetch the writer schema's latest
121/// version at `CREATE SOURCE` time).
122#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
123pub enum GlueAvroOptionName {
124    /// The `SCHEMA NAME [=] '<name>'` option. Names the schema within a
125    /// Glue registry whose latest version is fetched during purification.
126    SchemaName,
127}
128
129impl AstDisplay for GlueAvroOptionName {
130    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
131        match self {
132            GlueAvroOptionName::SchemaName => f.write_str("SCHEMA NAME"),
133        }
134    }
135}
136
137impl WithOptionName for GlueAvroOptionName {
138    /// # WARNING
139    ///
140    /// Whenever implementing this trait consider very carefully whether or not
141    /// this value could contain sensitive user data. If you're uncertain, err
142    /// on the conservative side and return `true`.
143    fn redact_value(&self) -> bool {
144        match self {
145            // A schema *name* is a user-chosen identifier, no more
146            // sensitive than a table name.
147            Self::SchemaName => false,
148        }
149    }
150}
151
152#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
153pub struct GlueAvroOption<T: AstInfo> {
154    pub name: GlueAvroOptionName,
155    pub value: Option<WithOptionValue<T>>,
156}
157impl_display_for_with_option!(GlueAvroOption);
158
159#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
160pub struct AvroSchemaOption<T: AstInfo> {
161    pub name: AvroSchemaOptionName,
162    pub value: Option<WithOptionValue<T>>,
163}
164impl_display_for_with_option!(AvroSchemaOption);
165
166#[derive(Debug, Clone, PartialEq, Eq, Hash)]
167pub enum AvroSchema<T: AstInfo> {
168    Csr {
169        csr_connection: CsrConnectionAvro<T>,
170    },
171    InlineSchema {
172        schema: Schema,
173        with_options: Vec<AvroSchemaOption<T>>,
174    },
175    /// `USING AWS GLUE SCHEMA REGISTRY CONNECTION <name> (SCHEMA NAME = '<n>')`.
176    ///
177    /// Parallel to the `Csr` variant.
178    Glue {
179        connection: T::ItemName,
180        with_options: Vec<GlueAvroOption<T>>,
181        /// Normally populated during purification by fetching the named
182        /// schema's latest version from AWS Glue. The grammar also accepts a
183        /// user-written `SEED VALUE SCHEMA`, so this may be set on input; it is
184        /// not the intended authoring path, but it is not rejected.
185        seed: Option<GlueAvroSeed>,
186    },
187}
188
189impl<T: AstInfo> AstDisplay for AvroSchema<T> {
190    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
191        match self {
192            Self::Csr { csr_connection } => {
193                f.write_node(csr_connection);
194            }
195            Self::InlineSchema {
196                schema,
197                with_options,
198            } => {
199                f.write_str("USING ");
200                schema.fmt(f);
201                if !with_options.is_empty() {
202                    f.write_str(" (");
203                    f.write_node(&display::comma_separated(with_options));
204                    f.write_str(")");
205                }
206            }
207            Self::Glue {
208                connection,
209                with_options,
210                seed,
211            } => {
212                f.write_str("USING AWS GLUE SCHEMA REGISTRY CONNECTION ");
213                f.write_node(connection);
214                if !with_options.is_empty() {
215                    f.write_str(" (");
216                    f.write_node(&display::comma_separated(with_options));
217                    f.write_str(")");
218                }
219                if let Some(seed) = seed {
220                    f.write_str(" ");
221                    f.write_node(seed);
222                }
223            }
224        }
225    }
226}
227impl_display_t!(AvroSchema);
228
229#[derive(Debug, Clone, PartialEq, Eq, Hash)]
230pub enum ProtobufSchema<T: AstInfo> {
231    Csr {
232        csr_connection: CsrConnectionProtobuf<T>,
233    },
234    InlineSchema {
235        message_name: String,
236        schema: Schema,
237    },
238}
239
240impl<T: AstInfo> AstDisplay for ProtobufSchema<T> {
241    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
242        match self {
243            Self::Csr { csr_connection } => {
244                f.write_node(csr_connection);
245            }
246            Self::InlineSchema {
247                message_name,
248                schema,
249            } => {
250                f.write_str("MESSAGE '");
251                f.write_node(&display::escape_single_quote_string(message_name));
252                f.write_str("' USING ");
253                f.write_str(schema);
254            }
255        }
256    }
257}
258impl_display_t!(ProtobufSchema);
259
260#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
261pub enum CsrConfigOptionName<T: AstInfo> {
262    AvroKeyFullname,
263    AvroValueFullname,
264    NullDefaults,
265    AvroDocOn(AvroDocOn<T>),
266    KeyCompatibilityLevel,
267    ValueCompatibilityLevel,
268}
269
270impl<T: AstInfo> WithOptionName for CsrConfigOptionName<T> {
271    /// # WARNING
272    ///
273    /// Whenever implementing this trait consider very carefully whether or not
274    /// this value could contain sensitive user data. If you're uncertain, err
275    /// on the conservative side and return `true`.
276    fn redact_value(&self) -> bool {
277        match self {
278            Self::AvroKeyFullname
279            | Self::AvroValueFullname
280            | Self::NullDefaults
281            | Self::AvroDocOn(_)
282            | Self::KeyCompatibilityLevel
283            | Self::ValueCompatibilityLevel => false,
284        }
285    }
286}
287
288#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
289pub struct AvroDocOn<T: AstInfo> {
290    pub identifier: DocOnIdentifier<T>,
291    pub for_schema: DocOnSchema,
292}
293#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
294pub enum DocOnSchema {
295    KeyOnly,
296    ValueOnly,
297    All,
298}
299
300#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
301pub enum DocOnIdentifier<T: AstInfo> {
302    Column(ColumnName<T>),
303    Type(T::ItemName),
304}
305
306impl<T: AstInfo> AstDisplay for AvroDocOn<T> {
307    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
308        match &self.for_schema {
309            DocOnSchema::KeyOnly => f.write_str("KEY "),
310            DocOnSchema::ValueOnly => f.write_str("VALUE "),
311            DocOnSchema::All => {}
312        }
313        match &self.identifier {
314            DocOnIdentifier::Column(name) => {
315                f.write_str("DOC ON COLUMN ");
316                f.write_node(name);
317            }
318            DocOnIdentifier::Type(name) => {
319                f.write_str("DOC ON TYPE ");
320                f.write_node(name);
321            }
322        }
323    }
324}
325impl_display_t!(AvroDocOn);
326
327impl<T: AstInfo> AstDisplay for CsrConfigOptionName<T> {
328    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
329        match self {
330            CsrConfigOptionName::AvroKeyFullname => f.write_str("AVRO KEY FULLNAME"),
331            CsrConfigOptionName::AvroValueFullname => f.write_str("AVRO VALUE FULLNAME"),
332            CsrConfigOptionName::NullDefaults => f.write_str("NULL DEFAULTS"),
333            CsrConfigOptionName::AvroDocOn(doc_on) => f.write_node(doc_on),
334            CsrConfigOptionName::KeyCompatibilityLevel => f.write_str("KEY COMPATIBILITY LEVEL"),
335            CsrConfigOptionName::ValueCompatibilityLevel => {
336                f.write_str("VALUE COMPATIBILITY LEVEL")
337            }
338        }
339    }
340}
341impl_display_t!(CsrConfigOptionName);
342
343#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
344/// An option in a `{FROM|INTO} CONNECTION ...` statement.
345pub struct CsrConfigOption<T: AstInfo> {
346    pub name: CsrConfigOptionName<T>,
347    pub value: Option<WithOptionValue<T>>,
348}
349impl_display_for_with_option!(CsrConfigOption);
350impl_display_t!(CsrConfigOption);
351
352#[derive(Debug, Clone, PartialEq, Eq, Hash)]
353pub struct CsrConnection<T: AstInfo> {
354    pub connection: T::ItemName,
355    pub options: Vec<CsrConfigOption<T>>,
356}
357
358impl<T: AstInfo> AstDisplay for CsrConnection<T> {
359    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
360        f.write_str("CONNECTION ");
361        f.write_node(&self.connection);
362        if !self.options.is_empty() {
363            f.write_str(" (");
364            f.write_node(&display::comma_separated(&self.options));
365            f.write_str(")");
366        }
367    }
368}
369impl_display_t!(CsrConnection);
370
371#[derive(Debug, Clone, PartialEq, Eq, Hash)]
372pub enum ReaderSchemaSelectionStrategy {
373    Latest,
374    Inline(String),
375    ById(i32),
376}
377
378impl Default for ReaderSchemaSelectionStrategy {
379    fn default() -> Self {
380        Self::Latest
381    }
382}
383
384#[derive(Debug, Clone, PartialEq, Eq, Hash)]
385pub struct CsrConnectionAvro<T: AstInfo> {
386    pub connection: CsrConnection<T>,
387    pub key_strategy: Option<ReaderSchemaSelectionStrategy>,
388    pub value_strategy: Option<ReaderSchemaSelectionStrategy>,
389    pub seed: Option<CsrSeedAvro>,
390}
391
392impl<T: AstInfo> AstDisplay for CsrConnectionAvro<T> {
393    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
394        f.write_str("USING CONFLUENT SCHEMA REGISTRY ");
395        f.write_node(&self.connection);
396        if let Some(seed) = &self.seed {
397            f.write_str(" ");
398            f.write_node(seed);
399        }
400    }
401}
402impl_display_t!(CsrConnectionAvro);
403
404#[derive(Debug, Clone, PartialEq, Eq, Hash)]
405pub struct CsrConnectionProtobuf<T: AstInfo> {
406    pub connection: CsrConnection<T>,
407    pub seed: Option<CsrSeedProtobuf>,
408}
409
410impl<T: AstInfo> AstDisplay for CsrConnectionProtobuf<T> {
411    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
412        f.write_str("USING CONFLUENT SCHEMA REGISTRY ");
413        f.write_node(&self.connection);
414
415        if let Some(seed) = &self.seed {
416            f.write_str(" ");
417            f.write_node(seed);
418        }
419    }
420}
421impl_display_t!(CsrConnectionProtobuf);
422
423#[derive(Debug, Clone, PartialEq, Eq, Hash)]
424pub struct CsrSeedAvro {
425    pub key_schema: Option<String>,
426    pub value_schema: String,
427    /// Reference schemas for the key schema, in dependency order.
428    /// Populated during purification by fetching from the schema registry.
429    pub key_reference_schemas: Vec<String>,
430    /// Reference schemas for the value schema, in dependency order.
431    /// Populated during purification by fetching from the schema registry.
432    pub value_reference_schemas: Vec<String>,
433}
434
435impl AstDisplay for CsrSeedAvro {
436    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
437        f.write_str("SEED");
438        if let Some(key_schema) = &self.key_schema {
439            f.write_str(" KEY SCHEMA '");
440            f.write_node(&display::escape_single_quote_string(key_schema));
441            f.write_str("'");
442            if !self.key_reference_schemas.is_empty() {
443                f.write_str(" KEY REFERENCES (");
444                for (i, schema) in self.key_reference_schemas.iter().enumerate() {
445                    if i > 0 {
446                        f.write_str(", ");
447                    }
448                    f.write_str("'");
449                    f.write_node(&display::escape_single_quote_string(schema));
450                    f.write_str("'");
451                }
452                f.write_str(")");
453            }
454        }
455        f.write_str(" VALUE SCHEMA '");
456        f.write_node(&display::escape_single_quote_string(&self.value_schema));
457        f.write_str("'");
458        if !self.value_reference_schemas.is_empty() {
459            f.write_str(" VALUE REFERENCES (");
460            for (i, schema) in self.value_reference_schemas.iter().enumerate() {
461                if i > 0 {
462                    f.write_str(", ");
463                }
464                f.write_str("'");
465                f.write_node(&display::escape_single_quote_string(schema));
466                f.write_str("'");
467            }
468            f.write_str(")");
469        }
470    }
471}
472impl_display!(CsrSeedAvro);
473
474/// Resolved reader schema for a single `AvroSchema::Glue` FORMAT clause.
475///
476/// Glue resolves exactly one named schema per FORMAT clause, so this holds a
477/// single schema rather than a key/value pair. This differs from
478/// [`CsrSeedAvro`], where one `FORMAT AVRO USING CONFLUENT ...` clause can seed
479/// both key and value. Under Glue, the key and value are expressed as separate
480/// `KEY FORMAT ... VALUE FORMAT ...` clauses, each carrying its own
481/// `GlueAvroSeed`; the field is named `value_schema` because, for a single
482/// clause, it is decoded as that clause's value (a bare `FORMAT` clause, or the
483/// value side of a `KEY FORMAT/VALUE FORMAT` pair) — and a `KEY FORMAT` clause
484/// reuses the same schema as its key.
485///
486/// Glue schemas also have no references (each schema-version is a single
487/// self-contained definition), so unlike [`CsrSeedAvro`] there is no references
488/// vector.
489#[derive(Debug, Clone, PartialEq, Eq, Hash)]
490pub struct GlueAvroSeed {
491    pub value_schema: String,
492}
493
494impl AstDisplay for GlueAvroSeed {
495    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
496        f.write_str("SEED VALUE SCHEMA '");
497        f.write_node(&display::escape_single_quote_string(&self.value_schema));
498        f.write_str("'");
499    }
500}
501impl_display!(GlueAvroSeed);
502
503#[derive(Debug, Clone, PartialEq, Eq, Hash)]
504pub struct CsrSeedProtobuf {
505    pub key: Option<CsrSeedProtobufSchema>,
506    pub value: CsrSeedProtobufSchema,
507}
508
509impl AstDisplay for CsrSeedProtobuf {
510    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
511        f.write_str("SEED");
512        if let Some(key) = &self.key {
513            f.write_str(" KEY ");
514            f.write_node(key);
515        }
516        f.write_str(" VALUE ");
517        f.write_node(&self.value);
518    }
519}
520impl_display!(CsrSeedProtobuf);
521
522#[derive(Debug, Clone, PartialEq, Eq, Hash)]
523pub struct CsrSeedProtobufSchema {
524    // Hex encoded string.
525    pub schema: String,
526    pub message_name: String,
527}
528impl AstDisplay for CsrSeedProtobufSchema {
529    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
530        f.write_str("SCHEMA '");
531        f.write_str(&display::escape_single_quote_string(&self.schema));
532        f.write_str("' MESSAGE '");
533        f.write_str(&display::escape_single_quote_string(&self.message_name));
534        f.write_str("'");
535    }
536}
537impl_display!(CsrSeedProtobufSchema);
538
539#[derive(Debug, Clone, PartialEq, Eq, Hash)]
540pub enum FormatSpecifier<T: AstInfo> {
541    /// `CREATE SOURCE/SINK .. FORMAT`
542    Bare(Format<T>),
543    /// `CREATE SOURCE/SINK .. KEY FORMAT .. VALUE FORMAT`
544    KeyValue { key: Format<T>, value: Format<T> },
545}
546
547impl<T: AstInfo> AstDisplay for FormatSpecifier<T> {
548    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
549        match self {
550            FormatSpecifier::Bare(format) => {
551                f.write_str("FORMAT ");
552                f.write_node(format)
553            }
554            FormatSpecifier::KeyValue { key, value } => {
555                f.write_str("KEY FORMAT ");
556                f.write_node(key);
557                f.write_str(" VALUE FORMAT ");
558                f.write_node(value);
559            }
560        }
561    }
562}
563impl_display_t!(FormatSpecifier);
564
565#[derive(Debug, Clone, PartialEq, Eq, Hash)]
566pub enum Format<T: AstInfo> {
567    Bytes,
568    Avro(AvroSchema<T>),
569    Protobuf(ProtobufSchema<T>),
570    Regex(String),
571    Csv {
572        columns: CsvColumns,
573        delimiter: char,
574    },
575    Json {
576        array: bool,
577    },
578    Text,
579}
580
581#[derive(Debug, Clone, PartialEq, Eq, Hash)]
582pub enum CsvColumns {
583    /// `WITH count COLUMNS`
584    Count(u64),
585    /// `WITH HEADER (ident, ...)?`: `names` is empty if there are no names specified
586    Header { names: Vec<Ident> },
587}
588
589impl AstDisplay for CsvColumns {
590    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
591        match self {
592            CsvColumns::Count(n) => {
593                f.write_str(n);
594                f.write_str(" COLUMNS")
595            }
596            CsvColumns::Header { names } => {
597                f.write_str("HEADER");
598                if !names.is_empty() {
599                    f.write_str(" (");
600                    f.write_node(&display::comma_separated(names));
601                    f.write_str(")");
602                }
603            }
604        }
605    }
606}
607
608#[derive(Debug, Clone, PartialEq, Eq, Hash)]
609pub enum SourceIncludeMetadata {
610    Key {
611        alias: Option<Ident>,
612    },
613    Timestamp {
614        alias: Option<Ident>,
615    },
616    Partition {
617        alias: Option<Ident>,
618    },
619    Offset {
620        alias: Option<Ident>,
621    },
622    Headers {
623        alias: Option<Ident>,
624    },
625    Header {
626        key: String,
627        alias: Ident,
628        use_bytes: bool,
629    },
630}
631
632impl AstDisplay for SourceIncludeMetadata {
633    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
634        let print_alias = |f: &mut AstFormatter<W>, alias: &Option<Ident>| {
635            if let Some(alias) = alias {
636                f.write_str(" AS ");
637                f.write_node(alias);
638            }
639        };
640
641        match self {
642            SourceIncludeMetadata::Key { alias } => {
643                f.write_str("KEY");
644                print_alias(f, alias);
645            }
646            SourceIncludeMetadata::Timestamp { alias } => {
647                f.write_str("TIMESTAMP");
648                print_alias(f, alias);
649            }
650            SourceIncludeMetadata::Partition { alias } => {
651                f.write_str("PARTITION");
652                print_alias(f, alias);
653            }
654            SourceIncludeMetadata::Offset { alias } => {
655                f.write_str("OFFSET");
656                print_alias(f, alias);
657            }
658            SourceIncludeMetadata::Headers { alias } => {
659                f.write_str("HEADERS");
660                print_alias(f, alias);
661            }
662            SourceIncludeMetadata::Header {
663                alias,
664                key,
665                use_bytes,
666            } => {
667                f.write_str("HEADER '");
668                f.write_str(&display::escape_single_quote_string(key));
669                f.write_str("'");
670                print_alias(f, &Some(alias.clone()));
671                if *use_bytes {
672                    f.write_str(" BYTES");
673                }
674            }
675        }
676    }
677}
678impl_display!(SourceIncludeMetadata);
679
680#[derive(Debug, Clone, PartialEq, Eq, Hash)]
681pub enum SourceErrorPolicy {
682    Inline {
683        /// The alias to use for the error column. If unspecified will be `error`.
684        alias: Option<Ident>,
685    },
686}
687
688impl AstDisplay for SourceErrorPolicy {
689    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
690        match self {
691            Self::Inline { alias } => {
692                f.write_str("INLINE");
693                if let Some(alias) = alias {
694                    f.write_str(" AS ");
695                    f.write_node(alias);
696                }
697            }
698        }
699    }
700}
701impl_display!(SourceErrorPolicy);
702
703#[derive(Debug, Clone, PartialEq, Eq, Hash)]
704pub enum SourceEnvelope {
705    None,
706    Debezium,
707    Upsert {
708        value_decode_err_policy: Vec<SourceErrorPolicy>,
709    },
710    CdcV2,
711}
712
713impl SourceEnvelope {
714    /// `true` iff Materialize is expected to crash or exhibit UB
715    /// when attempting to ingest data starting at an offset other than zero.
716    pub fn requires_all_input(&self) -> bool {
717        match self {
718            SourceEnvelope::None => false,
719            SourceEnvelope::Debezium => false,
720            SourceEnvelope::Upsert { .. } => false,
721            SourceEnvelope::CdcV2 => true,
722        }
723    }
724}
725
726impl AstDisplay for SourceEnvelope {
727    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
728        match self {
729            Self::None => {
730                // this is unreachable as long as the default is None, but include it in case we ever change that
731                f.write_str("NONE");
732            }
733            Self::Debezium => {
734                f.write_str("DEBEZIUM");
735            }
736            Self::Upsert {
737                value_decode_err_policy,
738            } => {
739                if value_decode_err_policy.is_empty() {
740                    f.write_str("UPSERT");
741                } else {
742                    f.write_str("UPSERT (VALUE DECODING ERRORS = (");
743                    f.write_node(&display::comma_separated(value_decode_err_policy));
744                    f.write_str("))")
745                }
746            }
747            Self::CdcV2 => {
748                f.write_str("MATERIALIZE");
749            }
750        }
751    }
752}
753impl_display!(SourceEnvelope);
754
755#[derive(Debug, Clone, PartialEq, Eq, Hash)]
756pub enum SinkEnvelope {
757    Debezium,
758    Upsert,
759}
760
761impl AstDisplay for SinkEnvelope {
762    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
763        match self {
764            Self::Upsert => {
765                f.write_str("UPSERT");
766            }
767            Self::Debezium => {
768                f.write_str("DEBEZIUM");
769            }
770        }
771    }
772}
773impl_display!(SinkEnvelope);
774
775#[derive(Debug, Clone, PartialEq, Eq, Hash)]
776pub enum IcebergSinkMode {
777    Upsert,
778    Append,
779}
780
781impl AstDisplay for IcebergSinkMode {
782    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
783        match self {
784            Self::Upsert => {
785                f.write_str("UPSERT");
786            }
787            Self::Append => {
788                f.write_str("APPEND");
789            }
790        }
791    }
792}
793impl_display!(IcebergSinkMode);
794
795#[derive(Debug, Clone, PartialEq, Eq, Hash)]
796pub enum SubscribeOutput<T: AstInfo> {
797    Diffs,
798    WithinTimestampOrderBy { order_by: Vec<OrderByExpr<T>> },
799    EnvelopeUpsert { key_columns: Vec<Ident> },
800    EnvelopeDebezium { key_columns: Vec<Ident> },
801}
802
803impl<T: AstInfo> AstDisplay for SubscribeOutput<T> {
804    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
805        match self {
806            Self::Diffs => {}
807            Self::WithinTimestampOrderBy { order_by } => {
808                f.write_str(" WITHIN TIMESTAMP ORDER BY ");
809                f.write_node(&display::comma_separated(order_by));
810            }
811            Self::EnvelopeUpsert { key_columns } => {
812                f.write_str(" ENVELOPE UPSERT (KEY (");
813                f.write_node(&display::comma_separated(key_columns));
814                f.write_str("))");
815            }
816            Self::EnvelopeDebezium { key_columns } => {
817                f.write_str(" ENVELOPE DEBEZIUM (KEY (");
818                f.write_node(&display::comma_separated(key_columns));
819                f.write_str("))");
820            }
821        }
822    }
823}
824impl_display_t!(SubscribeOutput);
825
826impl<T: AstInfo> AstDisplay for Format<T> {
827    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
828        match self {
829            Self::Bytes => f.write_str("BYTES"),
830            Self::Avro(inner) => {
831                f.write_str("AVRO ");
832                f.write_node(inner);
833            }
834            Self::Protobuf(inner) => {
835                f.write_str("PROTOBUF ");
836                f.write_node(inner);
837            }
838            Self::Regex(regex) => {
839                f.write_str("REGEX '");
840                f.write_node(&display::escape_single_quote_string(regex));
841                f.write_str("'");
842            }
843            Self::Csv { columns, delimiter } => {
844                f.write_str("CSV WITH ");
845                f.write_node(columns);
846
847                if *delimiter != ',' {
848                    f.write_str(" DELIMITED BY '");
849                    f.write_node(&display::escape_single_quote_string(&delimiter.to_string()));
850                    f.write_str("'");
851                }
852            }
853            Self::Json { array } => {
854                f.write_str("JSON");
855                if *array {
856                    f.write_str(" ARRAY");
857                }
858            }
859            Self::Text => f.write_str("TEXT"),
860        }
861    }
862}
863impl_display_t!(Format);
864
865// All connection options are bundled together to allow us to parse `ALTER
866// CONNECTION` without specifying the type of connection we're altering. Il faut
867// souffrir pour être belle.
868#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
869pub enum ConnectionOptionName {
870    AccessKeyId,
871    AssumeRoleArn,
872    AssumeRoleSessionName,
873    AvailabilityZones,
874    AwsConnection,
875    AwsPrivatelink,
876    Broker,
877    Brokers,
878    Credential,
879    Database,
880    Endpoint,
881    GcpConnection,
882    Host,
883    Password,
884    Port,
885    ProgressTopic,
886    ProgressTopicReplicationFactor,
887    PublicKey1,
888    PublicKey2,
889    Region,
890    Registry,
891    SaslMechanisms,
892    SaslPassword,
893    SaslUsername,
894    Scope,
895    SecretAccessKey,
896    SecurityProtocol,
897    ServiceAccountKey,
898    ServiceName,
899    SshTunnel,
900    SslCertificate,
901    SslCertificateAuthority,
902    SslKey,
903    SslMode,
904    SessionToken,
905    CatalogType,
906    Url,
907    User,
908    Warehouse,
909}
910
911impl ConnectionOptionName {
912    pub(crate) fn value_contains_sensitive_data(&self) -> bool {
913        match self {
914            ConnectionOptionName::AccessKeyId
915            | ConnectionOptionName::Credential
916            | ConnectionOptionName::Password
917            | ConnectionOptionName::SaslPassword
918            | ConnectionOptionName::SaslUsername
919            | ConnectionOptionName::SecretAccessKey
920            | ConnectionOptionName::ServiceAccountKey
921            | ConnectionOptionName::SessionToken
922            | ConnectionOptionName::SslCertificate
923            | ConnectionOptionName::SslCertificateAuthority
924            | ConnectionOptionName::SslKey
925            | ConnectionOptionName::User => true,
926            ConnectionOptionName::AssumeRoleArn
927            | ConnectionOptionName::AssumeRoleSessionName
928            | ConnectionOptionName::AvailabilityZones
929            | ConnectionOptionName::AwsConnection
930            | ConnectionOptionName::AwsPrivatelink
931            | ConnectionOptionName::Broker
932            | ConnectionOptionName::Brokers
933            | ConnectionOptionName::Database
934            | ConnectionOptionName::Endpoint
935            | ConnectionOptionName::GcpConnection
936            | ConnectionOptionName::Host
937            | ConnectionOptionName::Port
938            | ConnectionOptionName::ProgressTopic
939            | ConnectionOptionName::ProgressTopicReplicationFactor
940            | ConnectionOptionName::PublicKey1
941            | ConnectionOptionName::PublicKey2
942            | ConnectionOptionName::Region
943            | ConnectionOptionName::Registry
944            | ConnectionOptionName::SaslMechanisms
945            | ConnectionOptionName::Scope
946            | ConnectionOptionName::SecurityProtocol
947            | ConnectionOptionName::ServiceName
948            | ConnectionOptionName::SshTunnel
949            | ConnectionOptionName::SslMode
950            | ConnectionOptionName::CatalogType
951            | ConnectionOptionName::Url
952            | ConnectionOptionName::Warehouse => false,
953        }
954    }
955}
956
957impl AstDisplay for ConnectionOptionName {
958    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
959        f.write_str(match self {
960            ConnectionOptionName::AccessKeyId => "ACCESS KEY ID",
961            ConnectionOptionName::AvailabilityZones => "AVAILABILITY ZONES",
962            ConnectionOptionName::AwsConnection => "AWS CONNECTION",
963            ConnectionOptionName::AwsPrivatelink => "AWS PRIVATELINK",
964            ConnectionOptionName::Broker => "BROKER",
965            ConnectionOptionName::Brokers => "BROKERS",
966            ConnectionOptionName::Credential => "CREDENTIAL",
967            ConnectionOptionName::Database => "DATABASE",
968            ConnectionOptionName::Endpoint => "ENDPOINT",
969            ConnectionOptionName::GcpConnection => "GCP CONNECTION",
970            ConnectionOptionName::Host => "HOST",
971            ConnectionOptionName::Password => "PASSWORD",
972            ConnectionOptionName::Port => "PORT",
973            ConnectionOptionName::ProgressTopic => "PROGRESS TOPIC",
974            ConnectionOptionName::ProgressTopicReplicationFactor => {
975                "PROGRESS TOPIC REPLICATION FACTOR"
976            }
977            ConnectionOptionName::PublicKey1 => "PUBLIC KEY 1",
978            ConnectionOptionName::PublicKey2 => "PUBLIC KEY 2",
979            ConnectionOptionName::Region => "REGION",
980            ConnectionOptionName::Registry => "REGISTRY",
981            ConnectionOptionName::AssumeRoleArn => "ASSUME ROLE ARN",
982            ConnectionOptionName::AssumeRoleSessionName => "ASSUME ROLE SESSION NAME",
983            ConnectionOptionName::SaslMechanisms => "SASL MECHANISMS",
984            ConnectionOptionName::SaslPassword => "SASL PASSWORD",
985            ConnectionOptionName::SaslUsername => "SASL USERNAME",
986            ConnectionOptionName::Scope => "SCOPE",
987            ConnectionOptionName::SecurityProtocol => "SECURITY PROTOCOL",
988            ConnectionOptionName::SecretAccessKey => "SECRET ACCESS KEY",
989            ConnectionOptionName::ServiceAccountKey => "SERVICE ACCOUNT KEY",
990            ConnectionOptionName::ServiceName => "SERVICE NAME",
991            ConnectionOptionName::SshTunnel => "SSH TUNNEL",
992            ConnectionOptionName::SslCertificate => "SSL CERTIFICATE",
993            ConnectionOptionName::SslCertificateAuthority => "SSL CERTIFICATE AUTHORITY",
994            ConnectionOptionName::SslKey => "SSL KEY",
995            ConnectionOptionName::SslMode => "SSL MODE",
996            ConnectionOptionName::SessionToken => "SESSION TOKEN",
997            ConnectionOptionName::CatalogType => "CATALOG TYPE",
998            ConnectionOptionName::Url => "URL",
999            ConnectionOptionName::User => "USER",
1000            ConnectionOptionName::Warehouse => "WAREHOUSE",
1001        })
1002    }
1003}
1004impl_display!(ConnectionOptionName);
1005
1006impl WithOptionName for ConnectionOptionName {
1007    /// # WARNING
1008    ///
1009    /// Whenever implementing this trait consider very carefully whether or not
1010    /// this value could contain sensitive user data. If you're uncertain, err
1011    /// on the conservative side and return `true`.
1012    fn redact_value(&self) -> bool {
1013        // Credential-bearing options keep their values in redacted mode, so an
1014        // inline credential literal renders as `'<REDACTED>'`. A `SECRET`
1015        // reference is unaffected either way. It renders as its catalog name
1016        // via `WithOptionValue::Secret`.
1017        self.value_contains_sensitive_data()
1018    }
1019}
1020
1021#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1022/// An option in a `CREATE CONNECTION`.
1023pub struct ConnectionOption<T: AstInfo> {
1024    pub name: ConnectionOptionName,
1025    pub value: Option<WithOptionValue<T>>,
1026}
1027impl_display_for_with_option!(ConnectionOption);
1028impl_display_t!(ConnectionOption);
1029
1030#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
1031pub enum CreateConnectionType {
1032    Aws,
1033    AwsPrivatelink,
1034    GlueSchemaRegistry,
1035    Gcp,
1036    Kafka,
1037    Csr,
1038    Postgres,
1039    Ssh,
1040    SqlServer,
1041    MySql,
1042    IcebergCatalog,
1043}
1044
1045impl CreateConnectionType {
1046    pub fn as_str(&self) -> &'static str {
1047        match self {
1048            Self::Kafka => "kafka",
1049            Self::Csr => "confluent-schema-registry",
1050            Self::Postgres => "postgres",
1051            Self::Aws => "aws",
1052            Self::AwsPrivatelink => "aws-privatelink",
1053            Self::GlueSchemaRegistry => "glue-schema-registry",
1054            Self::Gcp => "gcp",
1055            Self::Ssh => "ssh-tunnel",
1056            Self::MySql => "mysql",
1057            Self::SqlServer => "sql-server",
1058            Self::IcebergCatalog => "iceberg-catalog",
1059        }
1060    }
1061}
1062
1063impl AstDisplay for CreateConnectionType {
1064    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1065        match self {
1066            Self::Kafka => {
1067                f.write_str("KAFKA");
1068            }
1069            Self::Csr => {
1070                f.write_str("CONFLUENT SCHEMA REGISTRY");
1071            }
1072            Self::Postgres => {
1073                f.write_str("POSTGRES");
1074            }
1075            Self::Aws => {
1076                f.write_str("AWS");
1077            }
1078            Self::AwsPrivatelink => {
1079                f.write_str("AWS PRIVATELINK");
1080            }
1081            Self::GlueSchemaRegistry => {
1082                f.write_str("AWS GLUE SCHEMA REGISTRY");
1083            }
1084            Self::Gcp => {
1085                f.write_str("GCP");
1086            }
1087            Self::Ssh => {
1088                f.write_str("SSH TUNNEL");
1089            }
1090            Self::SqlServer => {
1091                f.write_str("SQL SERVER");
1092            }
1093            Self::MySql => {
1094                f.write_str("MYSQL");
1095            }
1096            Self::IcebergCatalog => {
1097                f.write_str("ICEBERG CATALOG");
1098            }
1099        }
1100    }
1101}
1102impl_display!(CreateConnectionType);
1103
1104#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1105pub enum CreateConnectionOptionName {
1106    Validate,
1107}
1108
1109impl AstDisplay for CreateConnectionOptionName {
1110    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1111        f.write_str(match self {
1112            CreateConnectionOptionName::Validate => "VALIDATE",
1113        })
1114    }
1115}
1116impl_display!(CreateConnectionOptionName);
1117
1118impl WithOptionName for CreateConnectionOptionName {
1119    /// # WARNING
1120    ///
1121    /// Whenever implementing this trait consider very carefully whether or not
1122    /// this value could contain sensitive user data. If you're uncertain, err
1123    /// on the conservative side and return `true`.
1124    fn redact_value(&self) -> bool {
1125        match self {
1126            CreateConnectionOptionName::Validate => false,
1127        }
1128    }
1129}
1130
1131#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1132/// An option in a `CREATE CONNECTION...` statement.
1133pub struct CreateConnectionOption<T: AstInfo> {
1134    pub name: CreateConnectionOptionName,
1135    pub value: Option<WithOptionValue<T>>,
1136}
1137impl_display_for_with_option!(CreateConnectionOption);
1138impl_display_t!(CreateConnectionOption);
1139
1140#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1141pub enum KafkaSourceConfigOptionName {
1142    GroupIdPrefix,
1143    Topic,
1144    TopicMetadataRefreshInterval,
1145    StartTimestamp,
1146    StartOffset,
1147}
1148
1149impl AstDisplay for KafkaSourceConfigOptionName {
1150    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1151        f.write_str(match self {
1152            KafkaSourceConfigOptionName::GroupIdPrefix => "GROUP ID PREFIX",
1153            KafkaSourceConfigOptionName::Topic => "TOPIC",
1154            KafkaSourceConfigOptionName::TopicMetadataRefreshInterval => {
1155                "TOPIC METADATA REFRESH INTERVAL"
1156            }
1157            KafkaSourceConfigOptionName::StartOffset => "START OFFSET",
1158            KafkaSourceConfigOptionName::StartTimestamp => "START TIMESTAMP",
1159        })
1160    }
1161}
1162impl_display!(KafkaSourceConfigOptionName);
1163
1164impl WithOptionName for KafkaSourceConfigOptionName {
1165    /// # WARNING
1166    ///
1167    /// Whenever implementing this trait consider very carefully whether or not
1168    /// this value could contain sensitive user data. If you're uncertain, err
1169    /// on the conservative side and return `true`.
1170    fn redact_value(&self) -> bool {
1171        match self {
1172            KafkaSourceConfigOptionName::GroupIdPrefix
1173            | KafkaSourceConfigOptionName::Topic
1174            | KafkaSourceConfigOptionName::TopicMetadataRefreshInterval
1175            | KafkaSourceConfigOptionName::StartOffset
1176            | KafkaSourceConfigOptionName::StartTimestamp => false,
1177        }
1178    }
1179}
1180
1181#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1182pub struct KafkaSourceConfigOption<T: AstInfo> {
1183    pub name: KafkaSourceConfigOptionName,
1184    pub value: Option<WithOptionValue<T>>,
1185}
1186impl_display_for_with_option!(KafkaSourceConfigOption);
1187impl_display_t!(KafkaSourceConfigOption);
1188
1189#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1190pub enum KafkaSinkConfigOptionName {
1191    CompressionType,
1192    PartitionBy,
1193    ProgressGroupIdPrefix,
1194    Topic,
1195    TransactionalIdPrefix,
1196    LegacyIds,
1197    TopicConfig,
1198    TopicMetadataRefreshInterval,
1199    TopicPartitionCount,
1200    TopicReplicationFactor,
1201}
1202
1203impl AstDisplay for KafkaSinkConfigOptionName {
1204    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1205        f.write_str(match self {
1206            KafkaSinkConfigOptionName::CompressionType => "COMPRESSION TYPE",
1207            KafkaSinkConfigOptionName::PartitionBy => "PARTITION BY",
1208            KafkaSinkConfigOptionName::ProgressGroupIdPrefix => "PROGRESS GROUP ID PREFIX",
1209            KafkaSinkConfigOptionName::Topic => "TOPIC",
1210            KafkaSinkConfigOptionName::TransactionalIdPrefix => "TRANSACTIONAL ID PREFIX",
1211            KafkaSinkConfigOptionName::LegacyIds => "LEGACY IDS",
1212            KafkaSinkConfigOptionName::TopicConfig => "TOPIC CONFIG",
1213            KafkaSinkConfigOptionName::TopicMetadataRefreshInterval => {
1214                "TOPIC METADATA REFRESH INTERVAL"
1215            }
1216            KafkaSinkConfigOptionName::TopicPartitionCount => "TOPIC PARTITION COUNT",
1217            KafkaSinkConfigOptionName::TopicReplicationFactor => "TOPIC REPLICATION FACTOR",
1218        })
1219    }
1220}
1221impl_display!(KafkaSinkConfigOptionName);
1222
1223impl WithOptionName for KafkaSinkConfigOptionName {
1224    /// # WARNING
1225    ///
1226    /// Whenever implementing this trait consider very carefully whether or not
1227    /// this value could contain sensitive user data. If you're uncertain, err
1228    /// on the conservative side and return `true`.
1229    fn redact_value(&self) -> bool {
1230        match self {
1231            KafkaSinkConfigOptionName::CompressionType
1232            | KafkaSinkConfigOptionName::ProgressGroupIdPrefix
1233            | KafkaSinkConfigOptionName::Topic
1234            | KafkaSinkConfigOptionName::TopicMetadataRefreshInterval
1235            | KafkaSinkConfigOptionName::TransactionalIdPrefix
1236            | KafkaSinkConfigOptionName::LegacyIds
1237            | KafkaSinkConfigOptionName::TopicConfig
1238            | KafkaSinkConfigOptionName::TopicPartitionCount
1239            | KafkaSinkConfigOptionName::TopicReplicationFactor => false,
1240            KafkaSinkConfigOptionName::PartitionBy => true,
1241        }
1242    }
1243}
1244
1245#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1246pub struct KafkaSinkConfigOption<T: AstInfo> {
1247    pub name: KafkaSinkConfigOptionName,
1248    pub value: Option<WithOptionValue<T>>,
1249}
1250impl_display_for_with_option!(KafkaSinkConfigOption);
1251impl_display_t!(KafkaSinkConfigOption);
1252
1253#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1254pub enum IcebergSinkConfigOptionName {
1255    Namespace,
1256    Table,
1257}
1258
1259impl AstDisplay for IcebergSinkConfigOptionName {
1260    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1261        f.write_str(match self {
1262            IcebergSinkConfigOptionName::Namespace => "NAMESPACE",
1263            IcebergSinkConfigOptionName::Table => "TABLE",
1264        })
1265    }
1266}
1267impl_display!(IcebergSinkConfigOptionName);
1268
1269impl WithOptionName for IcebergSinkConfigOptionName {
1270    /// # WARNING
1271    ///
1272    /// Whenever implementing this trait consider very carefully whether or not
1273    /// this value could contain sensitive user data. If you're uncertain, err
1274    /// on the conservative side and return `true`.
1275    fn redact_value(&self) -> bool {
1276        match self {
1277            IcebergSinkConfigOptionName::Namespace | IcebergSinkConfigOptionName::Table => false,
1278        }
1279    }
1280}
1281
1282#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1283pub struct IcebergSinkConfigOption<T: AstInfo> {
1284    pub name: IcebergSinkConfigOptionName,
1285    pub value: Option<WithOptionValue<T>>,
1286}
1287impl_display_for_with_option!(IcebergSinkConfigOption);
1288impl_display_t!(IcebergSinkConfigOption);
1289
1290#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1291pub enum PgConfigOptionName {
1292    /// Hex encoded string of binary serialization of
1293    /// `mz_storage_types::sources::postgres::PostgresSourcePublicationDetails`
1294    Details,
1295    /// The name of the publication to sync
1296    Publication,
1297    /// Columns whose types you want to unconditionally format as text
1298    /// NOTE(roshan): This value is kept around to allow round-tripping a
1299    /// `CREATE SOURCE` statement while we still allow creating implicit
1300    /// subsources from `CREATE SOURCE`, but will be removed once
1301    /// fully deprecating that feature and forcing users to use explicit
1302    /// `CREATE TABLE .. FROM SOURCE` statements
1303    TextColumns,
1304    /// Columns you want to exclude
1305    /// NOTE: This value is kept around to allow round-tripping a
1306    /// `CREATE SOURCE` statement while we still allow creating implicit
1307    /// subsources from `CREATE SOURCE`, but will be removed once
1308    /// fully deprecating that feature and forcing users to use explicit
1309    /// `CREATE TABLE .. FROM SOURCE` statements
1310    ExcludeColumns,
1311}
1312
1313impl AstDisplay for PgConfigOptionName {
1314    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1315        f.write_str(match self {
1316            PgConfigOptionName::Details => "DETAILS",
1317            PgConfigOptionName::Publication => "PUBLICATION",
1318            PgConfigOptionName::TextColumns => "TEXT COLUMNS",
1319            PgConfigOptionName::ExcludeColumns => "EXCLUDE COLUMNS",
1320        })
1321    }
1322}
1323impl_display!(PgConfigOptionName);
1324
1325impl WithOptionName for PgConfigOptionName {
1326    /// # WARNING
1327    ///
1328    /// Whenever implementing this trait consider very carefully whether or not
1329    /// this value could contain sensitive user data. If you're uncertain, err
1330    /// on the conservative side and return `true`.
1331    fn redact_value(&self) -> bool {
1332        match self {
1333            PgConfigOptionName::Details
1334            | PgConfigOptionName::Publication
1335            | PgConfigOptionName::TextColumns
1336            | PgConfigOptionName::ExcludeColumns => false,
1337        }
1338    }
1339}
1340
1341#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1342/// An option in a `{FROM|INTO} CONNECTION ...` statement.
1343pub struct PgConfigOption<T: AstInfo> {
1344    pub name: PgConfigOptionName,
1345    pub value: Option<WithOptionValue<T>>,
1346}
1347impl_display_for_with_option!(PgConfigOption);
1348impl_display_t!(PgConfigOption);
1349
1350#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1351pub enum MySqlConfigOptionName {
1352    /// Hex encoded string of binary serialization of
1353    /// `mz_storage_types::sources::mysql::MySqlSourceDetails`
1354    Details,
1355    /// Columns whose types you want to unconditionally format as text
1356    /// NOTE(roshan): This value is kept around to allow round-tripping a
1357    /// `CREATE SOURCE` statement while we still allow creating implicit
1358    /// subsources from `CREATE SOURCE`, but will be removed once
1359    /// fully deprecating that feature and forcing users to use explicit
1360    /// `CREATE TABLE .. FROM SOURCE` statements
1361    TextColumns,
1362    /// Columns you want to exclude
1363    /// NOTE(roshan): This value is kept around to allow round-tripping a
1364    /// `CREATE SOURCE` statement while we still allow creating implicit
1365    /// subsources from `CREATE SOURCE`, but will be removed once
1366    /// fully deprecating that feature and forcing users to use explicit
1367    /// `CREATE TABLE .. FROM SOURCE` statements
1368    ExcludeColumns,
1369}
1370
1371impl AstDisplay for MySqlConfigOptionName {
1372    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1373        f.write_str(match self {
1374            MySqlConfigOptionName::Details => "DETAILS",
1375            MySqlConfigOptionName::TextColumns => "TEXT COLUMNS",
1376            MySqlConfigOptionName::ExcludeColumns => "EXCLUDE COLUMNS",
1377        })
1378    }
1379}
1380impl_display!(MySqlConfigOptionName);
1381
1382impl WithOptionName for MySqlConfigOptionName {
1383    /// # WARNING
1384    ///
1385    /// Whenever implementing this trait consider very carefully whether or not
1386    /// this value could contain sensitive user data. If you're uncertain, err
1387    /// on the conservative side and return `true`.
1388    fn redact_value(&self) -> bool {
1389        match self {
1390            MySqlConfigOptionName::Details
1391            | MySqlConfigOptionName::TextColumns
1392            | MySqlConfigOptionName::ExcludeColumns => false,
1393        }
1394    }
1395}
1396
1397#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1398/// An option in a `{FROM|INTO} CONNECTION ...` statement.
1399pub struct MySqlConfigOption<T: AstInfo> {
1400    pub name: MySqlConfigOptionName,
1401    pub value: Option<WithOptionValue<T>>,
1402}
1403impl_display_for_with_option!(MySqlConfigOption);
1404impl_display_t!(MySqlConfigOption);
1405
1406#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1407pub enum SqlServerConfigOptionName {
1408    /// Hex encoded string of binary serialization of
1409    /// `mz_storage_types::sources::sql_server::SqlServerSourceDetails`.
1410    Details,
1411    /// Columns whose types you want to unconditionally format as text.
1412    ///
1413    /// NOTE(roshan): This value is kept around to allow round-tripping a
1414    /// `CREATE SOURCE` statement while we still allow creating implicit
1415    /// subsources from `CREATE SOURCE`, but will be removed once
1416    /// fully deprecating that feature and forcing users to use explicit
1417    /// `CREATE TABLE .. FROM SOURCE` statements
1418    TextColumns,
1419    /// Columns you want to exclude.
1420    ///
1421    /// NOTE(roshan): This value is kept around to allow round-tripping a
1422    /// `CREATE SOURCE` statement while we still allow creating implicit
1423    /// subsources from `CREATE SOURCE`, but will be removed once
1424    /// fully deprecating that feature and forcing users to use explicit
1425    /// `CREATE TABLE .. FROM SOURCE` statements
1426    ExcludeColumns,
1427}
1428
1429impl AstDisplay for SqlServerConfigOptionName {
1430    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1431        f.write_str(match self {
1432            SqlServerConfigOptionName::Details => "DETAILS",
1433            SqlServerConfigOptionName::TextColumns => "TEXT COLUMNS",
1434            SqlServerConfigOptionName::ExcludeColumns => "EXCLUDE COLUMNS",
1435        })
1436    }
1437}
1438impl_display!(SqlServerConfigOptionName);
1439
1440impl WithOptionName for SqlServerConfigOptionName {
1441    /// # WARNING
1442    ///
1443    /// Whenever implementing this trait consider very carefully whether or not
1444    /// this value could contain sensitive user data. If you're uncertain, err
1445    /// on the conservative side and return `true`.
1446    fn redact_value(&self) -> bool {
1447        match self {
1448            SqlServerConfigOptionName::Details
1449            | SqlServerConfigOptionName::TextColumns
1450            | SqlServerConfigOptionName::ExcludeColumns => false,
1451        }
1452    }
1453}
1454
1455#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1456/// An option in a `{FROM|INTO} CONNECTION ...` statement.
1457pub struct SqlServerConfigOption<T: AstInfo> {
1458    pub name: SqlServerConfigOptionName,
1459    pub value: Option<WithOptionValue<T>>,
1460}
1461impl_display_for_with_option!(SqlServerConfigOption);
1462impl_display_t!(SqlServerConfigOption);
1463
1464#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1465pub enum CreateSourceConnection<T: AstInfo> {
1466    Kafka {
1467        connection: T::ItemName,
1468        options: Vec<KafkaSourceConfigOption<T>>,
1469    },
1470    Postgres {
1471        connection: T::ItemName,
1472        options: Vec<PgConfigOption<T>>,
1473    },
1474    SqlServer {
1475        connection: T::ItemName,
1476        options: Vec<SqlServerConfigOption<T>>,
1477    },
1478    MySql {
1479        connection: T::ItemName,
1480        options: Vec<MySqlConfigOption<T>>,
1481    },
1482    LoadGenerator {
1483        generator: LoadGenerator,
1484        options: Vec<LoadGeneratorOption<T>>,
1485    },
1486}
1487
1488impl<T: AstInfo> AstDisplay for CreateSourceConnection<T> {
1489    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1490        match self {
1491            CreateSourceConnection::Kafka {
1492                connection,
1493                options,
1494            } => {
1495                f.write_str("KAFKA CONNECTION ");
1496                f.write_node(connection);
1497                if !options.is_empty() {
1498                    f.write_str(" (");
1499                    f.write_node(&display::comma_separated(options));
1500                    f.write_str(")");
1501                }
1502            }
1503            CreateSourceConnection::Postgres {
1504                connection,
1505                options,
1506            } => {
1507                f.write_str("POSTGRES CONNECTION ");
1508                f.write_node(connection);
1509                if !options.is_empty() {
1510                    f.write_str(" (");
1511                    f.write_node(&display::comma_separated(options));
1512                    f.write_str(")");
1513                }
1514            }
1515            CreateSourceConnection::SqlServer {
1516                connection,
1517                options,
1518            } => {
1519                f.write_str("SQL SERVER CONNECTION ");
1520                f.write_node(connection);
1521                if !options.is_empty() {
1522                    f.write_str(" (");
1523                    f.write_node(&display::comma_separated(options));
1524                    f.write_str(")");
1525                }
1526            }
1527            CreateSourceConnection::MySql {
1528                connection,
1529                options,
1530            } => {
1531                f.write_str("MYSQL CONNECTION ");
1532                f.write_node(connection);
1533                if !options.is_empty() {
1534                    f.write_str(" (");
1535                    f.write_node(&display::comma_separated(options));
1536                    f.write_str(")");
1537                }
1538            }
1539            CreateSourceConnection::LoadGenerator { generator, options } => {
1540                f.write_str("LOAD GENERATOR ");
1541                f.write_node(generator);
1542                if !options.is_empty() {
1543                    f.write_str(" (");
1544                    f.write_node(&display::comma_separated(options));
1545                    f.write_str(")");
1546                }
1547            }
1548        }
1549    }
1550}
1551impl_display_t!(CreateSourceConnection);
1552
1553#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1554pub enum LoadGenerator {
1555    Clock,
1556    Counter,
1557    Marketing,
1558    Auction,
1559    Datums,
1560    Tpch,
1561    KeyValue,
1562}
1563
1564impl AstDisplay for LoadGenerator {
1565    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1566        match self {
1567            Self::Counter => f.write_str("COUNTER"),
1568            Self::Clock => f.write_str("CLOCK"),
1569            Self::Marketing => f.write_str("MARKETING"),
1570            Self::Auction => f.write_str("AUCTION"),
1571            Self::Datums => f.write_str("DATUMS"),
1572            Self::Tpch => f.write_str("TPCH"),
1573            Self::KeyValue => f.write_str("KEY VALUE"),
1574        }
1575    }
1576}
1577impl_display!(LoadGenerator);
1578
1579impl LoadGenerator {
1580    /// Corresponds with the same mapping on the `LoadGenerator` enum defined in
1581    /// src/storage-types/src/sources/load_generator.rs, but re-defined here for
1582    /// cases where we only have the AST representation. This can be removed once
1583    /// the `ast_rewrite_sources_to_tables` migration is removed.
1584    pub fn schema_name(&self) -> &'static str {
1585        match self {
1586            LoadGenerator::Counter => "counter",
1587            LoadGenerator::Clock => "clock",
1588            LoadGenerator::Marketing => "marketing",
1589            LoadGenerator::Auction => "auction",
1590            LoadGenerator::Datums => "datums",
1591            LoadGenerator::Tpch => "tpch",
1592            LoadGenerator::KeyValue => "key_value",
1593        }
1594    }
1595}
1596
1597#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1598pub enum LoadGeneratorOptionName {
1599    ScaleFactor,
1600    TickInterval,
1601    AsOf,
1602    UpTo,
1603    MaxCardinality,
1604    Keys,
1605    SnapshotRounds,
1606    TransactionalSnapshot,
1607    ValueSize,
1608    Seed,
1609    Partitions,
1610    BatchSize,
1611}
1612
1613impl AstDisplay for LoadGeneratorOptionName {
1614    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1615        f.write_str(match self {
1616            LoadGeneratorOptionName::ScaleFactor => "SCALE FACTOR",
1617            LoadGeneratorOptionName::TickInterval => "TICK INTERVAL",
1618            LoadGeneratorOptionName::AsOf => "AS OF",
1619            LoadGeneratorOptionName::UpTo => "UP TO",
1620            LoadGeneratorOptionName::MaxCardinality => "MAX CARDINALITY",
1621            LoadGeneratorOptionName::Keys => "KEYS",
1622            LoadGeneratorOptionName::SnapshotRounds => "SNAPSHOT ROUNDS",
1623            LoadGeneratorOptionName::TransactionalSnapshot => "TRANSACTIONAL SNAPSHOT",
1624            LoadGeneratorOptionName::ValueSize => "VALUE SIZE",
1625            LoadGeneratorOptionName::Seed => "SEED",
1626            LoadGeneratorOptionName::Partitions => "PARTITIONS",
1627            LoadGeneratorOptionName::BatchSize => "BATCH SIZE",
1628        })
1629    }
1630}
1631impl_display!(LoadGeneratorOptionName);
1632
1633impl WithOptionName for LoadGeneratorOptionName {
1634    /// # WARNING
1635    ///
1636    /// Whenever implementing this trait consider very carefully whether or not
1637    /// this value could contain sensitive user data. If you're uncertain, err
1638    /// on the conservative side and return `true`.
1639    fn redact_value(&self) -> bool {
1640        match self {
1641            LoadGeneratorOptionName::ScaleFactor
1642            | LoadGeneratorOptionName::TickInterval
1643            | LoadGeneratorOptionName::AsOf
1644            | LoadGeneratorOptionName::UpTo
1645            | LoadGeneratorOptionName::MaxCardinality
1646            | LoadGeneratorOptionName::Keys
1647            | LoadGeneratorOptionName::SnapshotRounds
1648            | LoadGeneratorOptionName::TransactionalSnapshot
1649            | LoadGeneratorOptionName::ValueSize
1650            | LoadGeneratorOptionName::Partitions
1651            | LoadGeneratorOptionName::BatchSize
1652            | LoadGeneratorOptionName::Seed => false,
1653        }
1654    }
1655}
1656
1657#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1658/// An option in a `CREATE CONNECTION...SSH`.
1659pub struct LoadGeneratorOption<T: AstInfo> {
1660    pub name: LoadGeneratorOptionName,
1661    pub value: Option<WithOptionValue<T>>,
1662}
1663impl_display_for_with_option!(LoadGeneratorOption);
1664impl_display_t!(LoadGeneratorOption);
1665
1666#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1667pub enum CreateSinkConnection<T: AstInfo> {
1668    Kafka {
1669        connection: T::ItemName,
1670        options: Vec<KafkaSinkConfigOption<T>>,
1671        key: Option<SinkKey>,
1672        headers: Option<Ident>,
1673    },
1674    Iceberg {
1675        catalog_connection: T::ItemName,
1676
1677        /// AWS creds for the storage layer.
1678        aws_connection: Option<T::ItemName>,
1679
1680        key: Option<SinkKey>,
1681        options: Vec<IcebergSinkConfigOption<T>>,
1682    },
1683}
1684
1685impl<T: AstInfo> AstDisplay for CreateSinkConnection<T> {
1686    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1687        match self {
1688            CreateSinkConnection::Kafka {
1689                connection,
1690                options,
1691                key,
1692                headers,
1693            } => {
1694                f.write_str("KAFKA CONNECTION ");
1695                f.write_node(connection);
1696                if !options.is_empty() {
1697                    f.write_str(" (");
1698                    f.write_node(&display::comma_separated(options));
1699                    f.write_str(")");
1700                }
1701                if let Some(key) = key.as_ref() {
1702                    f.write_str(" ");
1703                    f.write_node(key);
1704                }
1705                if let Some(headers) = headers {
1706                    f.write_str(" HEADERS ");
1707                    f.write_node(headers);
1708                }
1709            }
1710            CreateSinkConnection::Iceberg {
1711                catalog_connection,
1712                aws_connection,
1713                key,
1714                options,
1715            } => {
1716                f.write_str("ICEBERG CATALOG CONNECTION ");
1717                f.write_node(catalog_connection);
1718                if !options.is_empty() {
1719                    f.write_str(" (");
1720                    f.write_node(&display::comma_separated(options));
1721                    f.write_str(")");
1722                }
1723                if let Some(aws_connection) = aws_connection {
1724                    f.write_str(" USING AWS CONNECTION ");
1725                    f.write_node(aws_connection);
1726                }
1727                if let Some(key) = key.as_ref() {
1728                    f.write_str(" ");
1729                    f.write_node(key);
1730                }
1731            }
1732        }
1733    }
1734}
1735impl_display_t!(CreateSinkConnection);
1736
1737#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1738pub struct SinkKey {
1739    pub key_columns: Vec<Ident>,
1740    pub not_enforced: bool,
1741}
1742
1743impl AstDisplay for SinkKey {
1744    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1745        f.write_str("KEY (");
1746        f.write_node(&display::comma_separated(&self.key_columns));
1747        f.write_str(")");
1748        if self.not_enforced {
1749            f.write_str(" NOT ENFORCED");
1750        }
1751    }
1752}
1753
1754/// A table-level constraint, specified in a `CREATE TABLE` or an
1755/// `ALTER TABLE ADD <constraint>` statement.
1756#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1757pub enum TableConstraint<T: AstInfo> {
1758    /// `[ CONSTRAINT <name> ] { PRIMARY KEY | UNIQUE (NULLS NOT DISTINCT)? } (<columns>)`
1759    Unique {
1760        name: Option<Ident>,
1761        columns: Vec<Ident>,
1762        /// Whether this is a `PRIMARY KEY` or just a `UNIQUE` constraint
1763        is_primary: bool,
1764        // Where this constraint treats each NULL value as distinct; only available on `UNIQUE`
1765        // constraints.
1766        nulls_not_distinct: bool,
1767    },
1768    /// A referential integrity constraint (`[ CONSTRAINT <name> ] FOREIGN KEY (<columns>)
1769    /// REFERENCES <foreign_table> (<referred_columns>)`)
1770    ForeignKey {
1771        name: Option<Ident>,
1772        columns: Vec<Ident>,
1773        foreign_table: T::ItemName,
1774        referred_columns: Vec<Ident>,
1775    },
1776    /// `[ CONSTRAINT <name> ] CHECK (<expr>)`
1777    Check {
1778        name: Option<Ident>,
1779        expr: Box<Expr<T>>,
1780    },
1781}
1782
1783impl<T: AstInfo> AstDisplay for TableConstraint<T> {
1784    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1785        match self {
1786            TableConstraint::Unique {
1787                name,
1788                columns,
1789                is_primary,
1790                nulls_not_distinct,
1791            } => {
1792                f.write_node(&display_constraint_name(name));
1793                if *is_primary {
1794                    f.write_str("PRIMARY KEY ");
1795                } else {
1796                    f.write_str("UNIQUE ");
1797                    if *nulls_not_distinct {
1798                        f.write_str("NULLS NOT DISTINCT ");
1799                    }
1800                }
1801                f.write_str("(");
1802                f.write_node(&display::comma_separated(columns));
1803                f.write_str(")");
1804            }
1805            TableConstraint::ForeignKey {
1806                name,
1807                columns,
1808                foreign_table,
1809                referred_columns,
1810            } => {
1811                f.write_node(&display_constraint_name(name));
1812                f.write_str("FOREIGN KEY (");
1813                f.write_node(&display::comma_separated(columns));
1814                f.write_str(") REFERENCES ");
1815                f.write_node(foreign_table);
1816                f.write_str("(");
1817                f.write_node(&display::comma_separated(referred_columns));
1818                f.write_str(")");
1819            }
1820            TableConstraint::Check { name, expr } => {
1821                f.write_node(&display_constraint_name(name));
1822                f.write_str("CHECK (");
1823                f.write_node(&expr);
1824                f.write_str(")");
1825            }
1826        }
1827    }
1828}
1829impl_display_t!(TableConstraint);
1830
1831/// A key constraint, specified in a `CREATE SOURCE`.
1832#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1833pub enum KeyConstraint {
1834    // PRIMARY KEY (<columns>) NOT ENFORCED
1835    PrimaryKeyNotEnforced { columns: Vec<Ident> },
1836}
1837
1838impl AstDisplay for KeyConstraint {
1839    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1840        match self {
1841            KeyConstraint::PrimaryKeyNotEnforced { columns } => {
1842                f.write_str("PRIMARY KEY ");
1843                f.write_str("(");
1844                f.write_node(&display::comma_separated(columns));
1845                f.write_str(") ");
1846                f.write_str("NOT ENFORCED");
1847            }
1848        }
1849    }
1850}
1851impl_display!(KeyConstraint);
1852
1853#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1854pub enum CreateSourceOptionName {
1855    TimestampInterval,
1856    RetainHistory,
1857}
1858
1859impl AstDisplay for CreateSourceOptionName {
1860    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1861        f.write_str(match self {
1862            CreateSourceOptionName::TimestampInterval => "TIMESTAMP INTERVAL",
1863            CreateSourceOptionName::RetainHistory => "RETAIN HISTORY",
1864        })
1865    }
1866}
1867impl_display!(CreateSourceOptionName);
1868
1869impl WithOptionName for CreateSourceOptionName {
1870    /// # WARNING
1871    ///
1872    /// Whenever implementing this trait consider very carefully whether or not
1873    /// this value could contain sensitive user data. If you're uncertain, err
1874    /// on the conservative side and return `true`.
1875    fn redact_value(&self) -> bool {
1876        match self {
1877            CreateSourceOptionName::TimestampInterval | CreateSourceOptionName::RetainHistory => {
1878                false
1879            }
1880        }
1881    }
1882}
1883
1884#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1885/// An option in a `CREATE SOURCE...` statement.
1886pub struct CreateSourceOption<T: AstInfo> {
1887    pub name: CreateSourceOptionName,
1888    pub value: Option<WithOptionValue<T>>,
1889}
1890impl_display_for_with_option!(CreateSourceOption);
1891impl_display_t!(CreateSourceOption);
1892
1893/// SQL column definition
1894#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1895pub struct ColumnDef<T: AstInfo> {
1896    pub name: Ident,
1897    pub data_type: T::DataType,
1898    pub collation: Option<UnresolvedItemName>,
1899    pub options: Vec<ColumnOptionDef<T>>,
1900}
1901
1902impl<T: AstInfo> AstDisplay for ColumnDef<T> {
1903    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1904        f.write_node(&self.name);
1905        f.write_str(" ");
1906        f.write_node(&self.data_type);
1907        if let Some(collation) = &self.collation {
1908            f.write_str(" COLLATE ");
1909            f.write_node(collation);
1910        }
1911        for option in &self.options {
1912            f.write_str(" ");
1913            f.write_node(option);
1914        }
1915    }
1916}
1917impl_display_t!(ColumnDef);
1918
1919/// An optionally-named `ColumnOption`: `[ CONSTRAINT <name> ] <column-option>`.
1920///
1921/// Note that implementations are substantially more permissive than the ANSI
1922/// specification on what order column options can be presented in, and whether
1923/// they are allowed to be named. The specification distinguishes between
1924/// constraints (NOT NULL, UNIQUE, PRIMARY KEY, and CHECK), which can be named
1925/// and can appear in any order, and other options (DEFAULT, GENERATED), which
1926/// cannot be named and must appear in a fixed order. PostgreSQL, however,
1927/// allows preceding any option with `CONSTRAINT <name>`, even those that are
1928/// not really constraints, like NULL and DEFAULT. MSSQL is less permissive,
1929/// allowing DEFAULT, UNIQUE, PRIMARY KEY and CHECK to be named, but not NULL or
1930/// NOT NULL constraints (the last of which is in violation of the spec).
1931///
1932/// For maximum flexibility, we don't distinguish between constraint and
1933/// non-constraint options, lumping them all together under the umbrella of
1934/// "column options," and we allow any column option to be named.
1935#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1936pub struct ColumnOptionDef<T: AstInfo> {
1937    pub name: Option<Ident>,
1938    pub option: ColumnOption<T>,
1939}
1940
1941impl<T: AstInfo> AstDisplay for ColumnOptionDef<T> {
1942    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1943        f.write_node(&display_constraint_name(&self.name));
1944        f.write_node(&self.option);
1945    }
1946}
1947impl_display_t!(ColumnOptionDef);
1948
1949/// `ColumnOption`s are modifiers that follow a column definition in a `CREATE
1950/// TABLE` statement.
1951#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1952pub enum ColumnOption<T: AstInfo> {
1953    /// `NULL`
1954    Null,
1955    /// `NOT NULL`
1956    NotNull,
1957    /// `DEFAULT <restricted-expr>`
1958    Default(Expr<T>),
1959    /// `{ PRIMARY KEY | UNIQUE }`
1960    Unique { is_primary: bool },
1961    /// A referential integrity constraint (`[FOREIGN KEY REFERENCES
1962    /// <foreign_table> (<referred_columns>)`).
1963    ForeignKey {
1964        foreign_table: UnresolvedItemName,
1965        referred_columns: Vec<Ident>,
1966    },
1967    /// `CHECK (<expr>)`
1968    Check(Expr<T>),
1969    /// `VERSION <action> <version>`
1970    Versioned {
1971        action: ColumnVersioned,
1972        version: Version,
1973    },
1974}
1975
1976impl<T: AstInfo> AstDisplay for ColumnOption<T> {
1977    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1978        use ColumnOption::*;
1979        match self {
1980            Null => f.write_str("NULL"),
1981            NotNull => f.write_str("NOT NULL"),
1982            Default(expr) => {
1983                f.write_str("DEFAULT ");
1984                f.write_node(expr);
1985            }
1986            Unique { is_primary } => {
1987                if *is_primary {
1988                    f.write_str("PRIMARY KEY");
1989                } else {
1990                    f.write_str("UNIQUE");
1991                }
1992            }
1993            ForeignKey {
1994                foreign_table,
1995                referred_columns,
1996            } => {
1997                f.write_str("REFERENCES ");
1998                f.write_node(foreign_table);
1999                f.write_str(" (");
2000                f.write_node(&display::comma_separated(referred_columns));
2001                f.write_str(")");
2002            }
2003            Check(expr) => {
2004                f.write_str("CHECK (");
2005                f.write_node(expr);
2006                f.write_str(")");
2007            }
2008            Versioned { action, version } => {
2009                f.write_str("VERSION ");
2010                f.write_node(action);
2011                f.write_str(" ");
2012                f.write_node(version);
2013            }
2014        }
2015    }
2016}
2017impl_display_t!(ColumnOption);
2018
2019#[derive(Debug, Clone, PartialEq, Eq, Hash)]
2020pub enum ColumnVersioned {
2021    Added,
2022}
2023
2024impl AstDisplay for ColumnVersioned {
2025    fn fmt<W>(&self, f: &mut AstFormatter<W>)
2026    where
2027        W: fmt::Write,
2028    {
2029        match self {
2030            // TODO(alter_table): Support dropped columns.
2031            ColumnVersioned::Added => f.write_str("ADDED"),
2032        }
2033    }
2034}
2035impl_display!(ColumnVersioned);
2036
2037fn display_constraint_name<'a>(name: &'a Option<Ident>) -> impl AstDisplay + 'a {
2038    struct ConstraintName<'a>(&'a Option<Ident>);
2039    impl<'a> AstDisplay for ConstraintName<'a> {
2040        fn fmt<W>(&self, f: &mut AstFormatter<W>)
2041        where
2042            W: fmt::Write,
2043        {
2044            if let Some(name) = self.0 {
2045                f.write_str("CONSTRAINT ");
2046                f.write_node(name);
2047                f.write_str(" ");
2048            }
2049        }
2050    }
2051    ConstraintName(name)
2052}
mz_sql_parser/ast/defs/ddl.rs

mz_sql_parser/ast/defs/
ddl.rs
