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

mz_sql_parser/ast/defs/
ddl.rs
