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    AccessDelegation,
897    AccessKeyId,
898    AssumeRoleArn,
899    AssumeRoleSessionName,
900    AvailabilityZones,
901    AwsConnection,
902    AwsPrivatelink,
903    Broker,
904    Brokers,
905    Credential,
906    Database,
907    Endpoint,
908    GcpConnection,
909    Host,
910    Oauth2ServerUrl,
911    Password,
912    Port,
913    ProgressTopic,
914    ProgressTopicReplicationFactor,
915    PublicKey1,
916    PublicKey2,
917    Region,
918    Registry,
919    SaslMechanisms,
920    SaslPassword,
921    SaslUsername,
922    Scope,
923    SecretAccessKey,
924    SecurityProtocol,
925    ServiceAccountKey,
926    ServiceName,
927    SshTunnel,
928    SslCertificate,
929    SslCertificateAuthority,
930    SslKey,
931    SslMode,
932    SessionToken,
933    CatalogType,
934    StorageProvider,
935    Url,
936    User,
937    Warehouse,
938}
939
940impl ConnectionOptionName {
941    pub(crate) fn value_contains_sensitive_data(&self) -> bool {
942        match self {
943            ConnectionOptionName::AccessKeyId
944            | ConnectionOptionName::Credential
945            | ConnectionOptionName::Password
946            | ConnectionOptionName::SaslPassword
947            | ConnectionOptionName::SaslUsername
948            | ConnectionOptionName::SecretAccessKey
949            | ConnectionOptionName::ServiceAccountKey
950            | ConnectionOptionName::SessionToken
951            | ConnectionOptionName::SslCertificate
952            | ConnectionOptionName::SslCertificateAuthority
953            | ConnectionOptionName::SslKey
954            | ConnectionOptionName::User => true,
955            ConnectionOptionName::AccessDelegation
956            | ConnectionOptionName::AssumeRoleArn
957            | ConnectionOptionName::AssumeRoleSessionName
958            | ConnectionOptionName::AvailabilityZones
959            | ConnectionOptionName::AwsConnection
960            | ConnectionOptionName::AwsPrivatelink
961            | ConnectionOptionName::Broker
962            | ConnectionOptionName::Brokers
963            | ConnectionOptionName::Database
964            | ConnectionOptionName::Endpoint
965            | ConnectionOptionName::GcpConnection
966            | ConnectionOptionName::Host
967            | ConnectionOptionName::Oauth2ServerUrl
968            | ConnectionOptionName::Port
969            | ConnectionOptionName::ProgressTopic
970            | ConnectionOptionName::ProgressTopicReplicationFactor
971            | ConnectionOptionName::PublicKey1
972            | ConnectionOptionName::PublicKey2
973            | ConnectionOptionName::Region
974            | ConnectionOptionName::Registry
975            | ConnectionOptionName::SaslMechanisms
976            | ConnectionOptionName::Scope
977            | ConnectionOptionName::SecurityProtocol
978            | ConnectionOptionName::ServiceName
979            | ConnectionOptionName::SshTunnel
980            | ConnectionOptionName::SslMode
981            | ConnectionOptionName::CatalogType
982            | ConnectionOptionName::StorageProvider
983            | ConnectionOptionName::Url
984            | ConnectionOptionName::Warehouse => false,
985        }
986    }
987}
988
989impl AstDisplay for ConnectionOptionName {
990    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
991        f.write_str(match self {
992            ConnectionOptionName::AccessDelegation => "ACCESS DELEGATION",
993            ConnectionOptionName::AccessKeyId => "ACCESS KEY ID",
994            ConnectionOptionName::AvailabilityZones => "AVAILABILITY ZONES",
995            ConnectionOptionName::AwsConnection => "AWS CONNECTION",
996            ConnectionOptionName::AwsPrivatelink => "AWS PRIVATELINK",
997            ConnectionOptionName::Broker => "BROKER",
998            ConnectionOptionName::Brokers => "BROKERS",
999            ConnectionOptionName::Credential => "CREDENTIAL",
1000            ConnectionOptionName::Database => "DATABASE",
1001            ConnectionOptionName::Endpoint => "ENDPOINT",
1002            ConnectionOptionName::GcpConnection => "GCP CONNECTION",
1003            ConnectionOptionName::Host => "HOST",
1004            ConnectionOptionName::Oauth2ServerUrl => "OAUTH2 SERVER URL",
1005            ConnectionOptionName::Password => "PASSWORD",
1006            ConnectionOptionName::Port => "PORT",
1007            ConnectionOptionName::ProgressTopic => "PROGRESS TOPIC",
1008            ConnectionOptionName::ProgressTopicReplicationFactor => {
1009                "PROGRESS TOPIC REPLICATION FACTOR"
1010            }
1011            ConnectionOptionName::PublicKey1 => "PUBLIC KEY 1",
1012            ConnectionOptionName::PublicKey2 => "PUBLIC KEY 2",
1013            ConnectionOptionName::Region => "REGION",
1014            ConnectionOptionName::Registry => "REGISTRY",
1015            ConnectionOptionName::AssumeRoleArn => "ASSUME ROLE ARN",
1016            ConnectionOptionName::AssumeRoleSessionName => "ASSUME ROLE SESSION NAME",
1017            ConnectionOptionName::SaslMechanisms => "SASL MECHANISMS",
1018            ConnectionOptionName::SaslPassword => "SASL PASSWORD",
1019            ConnectionOptionName::SaslUsername => "SASL USERNAME",
1020            ConnectionOptionName::Scope => "SCOPE",
1021            ConnectionOptionName::SecurityProtocol => "SECURITY PROTOCOL",
1022            ConnectionOptionName::SecretAccessKey => "SECRET ACCESS KEY",
1023            ConnectionOptionName::ServiceAccountKey => "SERVICE ACCOUNT KEY",
1024            ConnectionOptionName::ServiceName => "SERVICE NAME",
1025            ConnectionOptionName::SshTunnel => "SSH TUNNEL",
1026            ConnectionOptionName::SslCertificate => "SSL CERTIFICATE",
1027            ConnectionOptionName::SslCertificateAuthority => "SSL CERTIFICATE AUTHORITY",
1028            ConnectionOptionName::SslKey => "SSL KEY",
1029            ConnectionOptionName::SslMode => "SSL MODE",
1030            ConnectionOptionName::SessionToken => "SESSION TOKEN",
1031            ConnectionOptionName::CatalogType => "CATALOG TYPE",
1032            ConnectionOptionName::StorageProvider => "STORAGE PROVIDER",
1033            ConnectionOptionName::Url => "URL",
1034            ConnectionOptionName::User => "USER",
1035            ConnectionOptionName::Warehouse => "WAREHOUSE",
1036        })
1037    }
1038}
1039impl_display!(ConnectionOptionName);
1040
1041impl WithOptionName for ConnectionOptionName {
1042    /// # WARNING
1043    ///
1044    /// Whenever implementing this trait consider very carefully whether or not
1045    /// this value could contain sensitive user data. If you're uncertain, err
1046    /// on the conservative side and return `true`.
1047    fn redact_value(&self) -> bool {
1048        // Credential-bearing options keep their values in redacted mode, so an
1049        // inline credential literal renders as `'<REDACTED>'`. A `SECRET`
1050        // reference is unaffected either way. It renders as its catalog name
1051        // via `WithOptionValue::Secret`.
1052        self.value_contains_sensitive_data()
1053    }
1054}
1055
1056#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1057/// An option in a `CREATE CONNECTION`.
1058pub struct ConnectionOption<T: AstInfo> {
1059    pub name: ConnectionOptionName,
1060    pub value: Option<WithOptionValue<T>>,
1061}
1062impl_display_for_with_option!(ConnectionOption);
1063impl_display_t!(ConnectionOption);
1064
1065#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
1066pub enum CreateConnectionType {
1067    Aws,
1068    AwsPrivatelink,
1069    GlueSchemaRegistry,
1070    Gcp,
1071    Kafka,
1072    Csr,
1073    Postgres,
1074    Ssh,
1075    SqlServer,
1076    MySql,
1077    IcebergCatalog,
1078}
1079
1080impl CreateConnectionType {
1081    pub fn as_str(&self) -> &'static str {
1082        match self {
1083            Self::Kafka => "kafka",
1084            Self::Csr => "confluent-schema-registry",
1085            Self::Postgres => "postgres",
1086            Self::Aws => "aws",
1087            Self::AwsPrivatelink => "aws-privatelink",
1088            Self::GlueSchemaRegistry => "glue-schema-registry",
1089            Self::Gcp => "gcp",
1090            Self::Ssh => "ssh-tunnel",
1091            Self::MySql => "mysql",
1092            Self::SqlServer => "sql-server",
1093            Self::IcebergCatalog => "iceberg-catalog",
1094        }
1095    }
1096}
1097
1098impl AstDisplay for CreateConnectionType {
1099    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1100        match self {
1101            Self::Kafka => {
1102                f.write_str("KAFKA");
1103            }
1104            Self::Csr => {
1105                f.write_str("CONFLUENT SCHEMA REGISTRY");
1106            }
1107            Self::Postgres => {
1108                f.write_str("POSTGRES");
1109            }
1110            Self::Aws => {
1111                f.write_str("AWS");
1112            }
1113            Self::AwsPrivatelink => {
1114                f.write_str("AWS PRIVATELINK");
1115            }
1116            Self::GlueSchemaRegistry => {
1117                f.write_str("AWS GLUE SCHEMA REGISTRY");
1118            }
1119            Self::Gcp => {
1120                f.write_str("GCP");
1121            }
1122            Self::Ssh => {
1123                f.write_str("SSH TUNNEL");
1124            }
1125            Self::SqlServer => {
1126                f.write_str("SQL SERVER");
1127            }
1128            Self::MySql => {
1129                f.write_str("MYSQL");
1130            }
1131            Self::IcebergCatalog => {
1132                f.write_str("ICEBERG CATALOG");
1133            }
1134        }
1135    }
1136}
1137impl_display!(CreateConnectionType);
1138
1139#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1140pub enum CreateConnectionOptionName {
1141    Validate,
1142}
1143
1144impl AstDisplay for CreateConnectionOptionName {
1145    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1146        f.write_str(match self {
1147            CreateConnectionOptionName::Validate => "VALIDATE",
1148        })
1149    }
1150}
1151impl_display!(CreateConnectionOptionName);
1152
1153impl WithOptionName for CreateConnectionOptionName {
1154    /// # WARNING
1155    ///
1156    /// Whenever implementing this trait consider very carefully whether or not
1157    /// this value could contain sensitive user data. If you're uncertain, err
1158    /// on the conservative side and return `true`.
1159    fn redact_value(&self) -> bool {
1160        match self {
1161            CreateConnectionOptionName::Validate => false,
1162        }
1163    }
1164}
1165
1166#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1167/// An option in a `CREATE CONNECTION...` statement.
1168pub struct CreateConnectionOption<T: AstInfo> {
1169    pub name: CreateConnectionOptionName,
1170    pub value: Option<WithOptionValue<T>>,
1171}
1172impl_display_for_with_option!(CreateConnectionOption);
1173impl_display_t!(CreateConnectionOption);
1174
1175#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1176pub enum KafkaSourceConfigOptionName {
1177    GroupIdPrefix,
1178    Topic,
1179    TopicMetadataRefreshInterval,
1180    StartTimestamp,
1181    StartOffset,
1182}
1183
1184impl AstDisplay for KafkaSourceConfigOptionName {
1185    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1186        f.write_str(match self {
1187            KafkaSourceConfigOptionName::GroupIdPrefix => "GROUP ID PREFIX",
1188            KafkaSourceConfigOptionName::Topic => "TOPIC",
1189            KafkaSourceConfigOptionName::TopicMetadataRefreshInterval => {
1190                "TOPIC METADATA REFRESH INTERVAL"
1191            }
1192            KafkaSourceConfigOptionName::StartOffset => "START OFFSET",
1193            KafkaSourceConfigOptionName::StartTimestamp => "START TIMESTAMP",
1194        })
1195    }
1196}
1197impl_display!(KafkaSourceConfigOptionName);
1198
1199impl WithOptionName for KafkaSourceConfigOptionName {
1200    /// # WARNING
1201    ///
1202    /// Whenever implementing this trait consider very carefully whether or not
1203    /// this value could contain sensitive user data. If you're uncertain, err
1204    /// on the conservative side and return `true`.
1205    fn redact_value(&self) -> bool {
1206        match self {
1207            KafkaSourceConfigOptionName::GroupIdPrefix
1208            | KafkaSourceConfigOptionName::Topic
1209            | KafkaSourceConfigOptionName::TopicMetadataRefreshInterval
1210            | KafkaSourceConfigOptionName::StartOffset
1211            | KafkaSourceConfigOptionName::StartTimestamp => false,
1212        }
1213    }
1214}
1215
1216#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1217pub struct KafkaSourceConfigOption<T: AstInfo> {
1218    pub name: KafkaSourceConfigOptionName,
1219    pub value: Option<WithOptionValue<T>>,
1220}
1221impl_display_for_with_option!(KafkaSourceConfigOption);
1222impl_display_t!(KafkaSourceConfigOption);
1223
1224#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1225pub enum KafkaSinkConfigOptionName {
1226    CompressionType,
1227    PartitionBy,
1228    ProgressGroupIdPrefix,
1229    Topic,
1230    TransactionalIdPrefix,
1231    LegacyIds,
1232    TopicConfig,
1233    TopicMetadataRefreshInterval,
1234    TopicPartitionCount,
1235    TopicReplicationFactor,
1236}
1237
1238impl AstDisplay for KafkaSinkConfigOptionName {
1239    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1240        f.write_str(match self {
1241            KafkaSinkConfigOptionName::CompressionType => "COMPRESSION TYPE",
1242            KafkaSinkConfigOptionName::PartitionBy => "PARTITION BY",
1243            KafkaSinkConfigOptionName::ProgressGroupIdPrefix => "PROGRESS GROUP ID PREFIX",
1244            KafkaSinkConfigOptionName::Topic => "TOPIC",
1245            KafkaSinkConfigOptionName::TransactionalIdPrefix => "TRANSACTIONAL ID PREFIX",
1246            KafkaSinkConfigOptionName::LegacyIds => "LEGACY IDS",
1247            KafkaSinkConfigOptionName::TopicConfig => "TOPIC CONFIG",
1248            KafkaSinkConfigOptionName::TopicMetadataRefreshInterval => {
1249                "TOPIC METADATA REFRESH INTERVAL"
1250            }
1251            KafkaSinkConfigOptionName::TopicPartitionCount => "TOPIC PARTITION COUNT",
1252            KafkaSinkConfigOptionName::TopicReplicationFactor => "TOPIC REPLICATION FACTOR",
1253        })
1254    }
1255}
1256impl_display!(KafkaSinkConfigOptionName);
1257
1258impl WithOptionName for KafkaSinkConfigOptionName {
1259    /// # WARNING
1260    ///
1261    /// Whenever implementing this trait consider very carefully whether or not
1262    /// this value could contain sensitive user data. If you're uncertain, err
1263    /// on the conservative side and return `true`.
1264    fn redact_value(&self) -> bool {
1265        match self {
1266            KafkaSinkConfigOptionName::CompressionType
1267            | KafkaSinkConfigOptionName::ProgressGroupIdPrefix
1268            | KafkaSinkConfigOptionName::Topic
1269            | KafkaSinkConfigOptionName::TopicMetadataRefreshInterval
1270            | KafkaSinkConfigOptionName::TransactionalIdPrefix
1271            | KafkaSinkConfigOptionName::LegacyIds
1272            | KafkaSinkConfigOptionName::TopicConfig
1273            | KafkaSinkConfigOptionName::TopicPartitionCount
1274            | KafkaSinkConfigOptionName::TopicReplicationFactor => false,
1275            KafkaSinkConfigOptionName::PartitionBy => true,
1276        }
1277    }
1278}
1279
1280#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1281pub struct KafkaSinkConfigOption<T: AstInfo> {
1282    pub name: KafkaSinkConfigOptionName,
1283    pub value: Option<WithOptionValue<T>>,
1284}
1285impl_display_for_with_option!(KafkaSinkConfigOption);
1286impl_display_t!(KafkaSinkConfigOption);
1287
1288#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1289pub enum IcebergSinkConfigOptionName {
1290    Namespace,
1291    Table,
1292}
1293
1294impl AstDisplay for IcebergSinkConfigOptionName {
1295    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1296        f.write_str(match self {
1297            IcebergSinkConfigOptionName::Namespace => "NAMESPACE",
1298            IcebergSinkConfigOptionName::Table => "TABLE",
1299        })
1300    }
1301}
1302impl_display!(IcebergSinkConfigOptionName);
1303
1304impl WithOptionName for IcebergSinkConfigOptionName {
1305    /// # WARNING
1306    ///
1307    /// Whenever implementing this trait consider very carefully whether or not
1308    /// this value could contain sensitive user data. If you're uncertain, err
1309    /// on the conservative side and return `true`.
1310    fn redact_value(&self) -> bool {
1311        match self {
1312            IcebergSinkConfigOptionName::Namespace | IcebergSinkConfigOptionName::Table => false,
1313        }
1314    }
1315}
1316
1317#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1318pub struct IcebergSinkConfigOption<T: AstInfo> {
1319    pub name: IcebergSinkConfigOptionName,
1320    pub value: Option<WithOptionValue<T>>,
1321}
1322impl_display_for_with_option!(IcebergSinkConfigOption);
1323impl_display_t!(IcebergSinkConfigOption);
1324
1325#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1326pub enum PgConfigOptionName {
1327    /// Hex encoded string of binary serialization of
1328    /// `mz_storage_types::sources::postgres::PostgresSourcePublicationDetails`
1329    Details,
1330    /// The name of the publication to sync
1331    Publication,
1332    /// Columns whose types you want to unconditionally format as text
1333    /// NOTE(roshan): This value is kept around to allow round-tripping a
1334    /// `CREATE SOURCE` statement while we still allow creating implicit
1335    /// subsources from `CREATE SOURCE`, but will be removed once
1336    /// fully deprecating that feature and forcing users to use explicit
1337    /// `CREATE TABLE .. FROM SOURCE` statements
1338    TextColumns,
1339    /// Columns you want to exclude
1340    /// NOTE: This value is kept around to allow round-tripping a
1341    /// `CREATE SOURCE` statement while we still allow creating implicit
1342    /// subsources from `CREATE SOURCE`, but will be removed once
1343    /// fully deprecating that feature and forcing users to use explicit
1344    /// `CREATE TABLE .. FROM SOURCE` statements
1345    ExcludeColumns,
1346}
1347
1348impl AstDisplay for PgConfigOptionName {
1349    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1350        f.write_str(match self {
1351            PgConfigOptionName::Details => "DETAILS",
1352            PgConfigOptionName::Publication => "PUBLICATION",
1353            PgConfigOptionName::TextColumns => "TEXT COLUMNS",
1354            PgConfigOptionName::ExcludeColumns => "EXCLUDE COLUMNS",
1355        })
1356    }
1357}
1358impl_display!(PgConfigOptionName);
1359
1360impl WithOptionName for PgConfigOptionName {
1361    /// # WARNING
1362    ///
1363    /// Whenever implementing this trait consider very carefully whether or not
1364    /// this value could contain sensitive user data. If you're uncertain, err
1365    /// on the conservative side and return `true`.
1366    fn redact_value(&self) -> bool {
1367        match self {
1368            PgConfigOptionName::Details
1369            | PgConfigOptionName::Publication
1370            | PgConfigOptionName::TextColumns
1371            | PgConfigOptionName::ExcludeColumns => false,
1372        }
1373    }
1374}
1375
1376#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1377/// An option in a `{FROM|INTO} CONNECTION ...` statement.
1378pub struct PgConfigOption<T: AstInfo> {
1379    pub name: PgConfigOptionName,
1380    pub value: Option<WithOptionValue<T>>,
1381}
1382impl_display_for_with_option!(PgConfigOption);
1383impl_display_t!(PgConfigOption);
1384
1385#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1386pub enum MySqlConfigOptionName {
1387    /// Hex encoded string of binary serialization of
1388    /// `mz_storage_types::sources::mysql::MySqlSourceDetails`
1389    Details,
1390    /// Columns whose types you want to unconditionally format as text
1391    /// NOTE(roshan): This value is kept around to allow round-tripping a
1392    /// `CREATE SOURCE` statement while we still allow creating implicit
1393    /// subsources from `CREATE SOURCE`, but will be removed once
1394    /// fully deprecating that feature and forcing users to use explicit
1395    /// `CREATE TABLE .. FROM SOURCE` statements
1396    TextColumns,
1397    /// Columns you want to exclude
1398    /// NOTE(roshan): This value is kept around to allow round-tripping a
1399    /// `CREATE SOURCE` statement while we still allow creating implicit
1400    /// subsources from `CREATE SOURCE`, but will be removed once
1401    /// fully deprecating that feature and forcing users to use explicit
1402    /// `CREATE TABLE .. FROM SOURCE` statements
1403    ExcludeColumns,
1404}
1405
1406impl AstDisplay for MySqlConfigOptionName {
1407    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1408        f.write_str(match self {
1409            MySqlConfigOptionName::Details => "DETAILS",
1410            MySqlConfigOptionName::TextColumns => "TEXT COLUMNS",
1411            MySqlConfigOptionName::ExcludeColumns => "EXCLUDE COLUMNS",
1412        })
1413    }
1414}
1415impl_display!(MySqlConfigOptionName);
1416
1417impl WithOptionName for MySqlConfigOptionName {
1418    /// # WARNING
1419    ///
1420    /// Whenever implementing this trait consider very carefully whether or not
1421    /// this value could contain sensitive user data. If you're uncertain, err
1422    /// on the conservative side and return `true`.
1423    fn redact_value(&self) -> bool {
1424        match self {
1425            MySqlConfigOptionName::Details
1426            | MySqlConfigOptionName::TextColumns
1427            | MySqlConfigOptionName::ExcludeColumns => false,
1428        }
1429    }
1430}
1431
1432#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1433/// An option in a `{FROM|INTO} CONNECTION ...` statement.
1434pub struct MySqlConfigOption<T: AstInfo> {
1435    pub name: MySqlConfigOptionName,
1436    pub value: Option<WithOptionValue<T>>,
1437}
1438impl_display_for_with_option!(MySqlConfigOption);
1439impl_display_t!(MySqlConfigOption);
1440
1441#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1442pub enum SqlServerConfigOptionName {
1443    /// Hex encoded string of binary serialization of
1444    /// `mz_storage_types::sources::sql_server::SqlServerSourceDetails`.
1445    Details,
1446    /// Columns whose types you want to unconditionally format as text.
1447    ///
1448    /// NOTE(roshan): This value is kept around to allow round-tripping a
1449    /// `CREATE SOURCE` statement while we still allow creating implicit
1450    /// subsources from `CREATE SOURCE`, but will be removed once
1451    /// fully deprecating that feature and forcing users to use explicit
1452    /// `CREATE TABLE .. FROM SOURCE` statements
1453    TextColumns,
1454    /// Columns you want to exclude.
1455    ///
1456    /// NOTE(roshan): This value is kept around to allow round-tripping a
1457    /// `CREATE SOURCE` statement while we still allow creating implicit
1458    /// subsources from `CREATE SOURCE`, but will be removed once
1459    /// fully deprecating that feature and forcing users to use explicit
1460    /// `CREATE TABLE .. FROM SOURCE` statements
1461    ExcludeColumns,
1462}
1463
1464impl AstDisplay for SqlServerConfigOptionName {
1465    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1466        f.write_str(match self {
1467            SqlServerConfigOptionName::Details => "DETAILS",
1468            SqlServerConfigOptionName::TextColumns => "TEXT COLUMNS",
1469            SqlServerConfigOptionName::ExcludeColumns => "EXCLUDE COLUMNS",
1470        })
1471    }
1472}
1473impl_display!(SqlServerConfigOptionName);
1474
1475impl WithOptionName for SqlServerConfigOptionName {
1476    /// # WARNING
1477    ///
1478    /// Whenever implementing this trait consider very carefully whether or not
1479    /// this value could contain sensitive user data. If you're uncertain, err
1480    /// on the conservative side and return `true`.
1481    fn redact_value(&self) -> bool {
1482        match self {
1483            SqlServerConfigOptionName::Details
1484            | SqlServerConfigOptionName::TextColumns
1485            | SqlServerConfigOptionName::ExcludeColumns => false,
1486        }
1487    }
1488}
1489
1490#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1491/// An option in a `{FROM|INTO} CONNECTION ...` statement.
1492pub struct SqlServerConfigOption<T: AstInfo> {
1493    pub name: SqlServerConfigOptionName,
1494    pub value: Option<WithOptionValue<T>>,
1495}
1496impl_display_for_with_option!(SqlServerConfigOption);
1497impl_display_t!(SqlServerConfigOption);
1498
1499#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1500pub enum CreateSourceConnection<T: AstInfo> {
1501    Kafka {
1502        connection: T::ItemName,
1503        options: Vec<KafkaSourceConfigOption<T>>,
1504    },
1505    Postgres {
1506        connection: T::ItemName,
1507        options: Vec<PgConfigOption<T>>,
1508    },
1509    SqlServer {
1510        connection: T::ItemName,
1511        options: Vec<SqlServerConfigOption<T>>,
1512    },
1513    MySql {
1514        connection: T::ItemName,
1515        options: Vec<MySqlConfigOption<T>>,
1516    },
1517    LoadGenerator {
1518        generator: LoadGenerator,
1519        options: Vec<LoadGeneratorOption<T>>,
1520    },
1521}
1522
1523impl<T: AstInfo> AstDisplay for CreateSourceConnection<T> {
1524    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1525        match self {
1526            CreateSourceConnection::Kafka {
1527                connection,
1528                options,
1529            } => {
1530                f.write_str("KAFKA CONNECTION ");
1531                f.write_node(connection);
1532                if !options.is_empty() {
1533                    f.write_str(" (");
1534                    f.write_node(&display::comma_separated(options));
1535                    f.write_str(")");
1536                }
1537            }
1538            CreateSourceConnection::Postgres {
1539                connection,
1540                options,
1541            } => {
1542                f.write_str("POSTGRES CONNECTION ");
1543                f.write_node(connection);
1544                if !options.is_empty() {
1545                    f.write_str(" (");
1546                    f.write_node(&display::comma_separated(options));
1547                    f.write_str(")");
1548                }
1549            }
1550            CreateSourceConnection::SqlServer {
1551                connection,
1552                options,
1553            } => {
1554                f.write_str("SQL SERVER CONNECTION ");
1555                f.write_node(connection);
1556                if !options.is_empty() {
1557                    f.write_str(" (");
1558                    f.write_node(&display::comma_separated(options));
1559                    f.write_str(")");
1560                }
1561            }
1562            CreateSourceConnection::MySql {
1563                connection,
1564                options,
1565            } => {
1566                f.write_str("MYSQL CONNECTION ");
1567                f.write_node(connection);
1568                if !options.is_empty() {
1569                    f.write_str(" (");
1570                    f.write_node(&display::comma_separated(options));
1571                    f.write_str(")");
1572                }
1573            }
1574            CreateSourceConnection::LoadGenerator { generator, options } => {
1575                f.write_str("LOAD GENERATOR ");
1576                f.write_node(generator);
1577                if !options.is_empty() {
1578                    f.write_str(" (");
1579                    f.write_node(&display::comma_separated(options));
1580                    f.write_str(")");
1581                }
1582            }
1583        }
1584    }
1585}
1586impl_display_t!(CreateSourceConnection);
1587
1588#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1589pub enum LoadGenerator {
1590    Clock,
1591    Counter,
1592    Marketing,
1593    Auction,
1594    Datums,
1595    Tpch,
1596    KeyValue,
1597}
1598
1599impl AstDisplay for LoadGenerator {
1600    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1601        match self {
1602            Self::Counter => f.write_str("COUNTER"),
1603            Self::Clock => f.write_str("CLOCK"),
1604            Self::Marketing => f.write_str("MARKETING"),
1605            Self::Auction => f.write_str("AUCTION"),
1606            Self::Datums => f.write_str("DATUMS"),
1607            Self::Tpch => f.write_str("TPCH"),
1608            Self::KeyValue => f.write_str("KEY VALUE"),
1609        }
1610    }
1611}
1612impl_display!(LoadGenerator);
1613
1614impl LoadGenerator {
1615    /// Corresponds with the same mapping on the `LoadGenerator` enum defined in
1616    /// src/storage-types/src/sources/load_generator.rs, but re-defined here for
1617    /// cases where we only have the AST representation. This can be removed once
1618    /// the `ast_rewrite_sources_to_tables` migration is removed.
1619    pub fn schema_name(&self) -> &'static str {
1620        match self {
1621            LoadGenerator::Counter => "counter",
1622            LoadGenerator::Clock => "clock",
1623            LoadGenerator::Marketing => "marketing",
1624            LoadGenerator::Auction => "auction",
1625            LoadGenerator::Datums => "datums",
1626            LoadGenerator::Tpch => "tpch",
1627            LoadGenerator::KeyValue => "key_value",
1628        }
1629    }
1630}
1631
1632#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1633pub enum LoadGeneratorOptionName {
1634    ScaleFactor,
1635    TickInterval,
1636    AsOf,
1637    UpTo,
1638    MaxCardinality,
1639    Keys,
1640    SnapshotRounds,
1641    TransactionalSnapshot,
1642    ValueSize,
1643    Seed,
1644    Partitions,
1645    BatchSize,
1646}
1647
1648impl AstDisplay for LoadGeneratorOptionName {
1649    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1650        f.write_str(match self {
1651            LoadGeneratorOptionName::ScaleFactor => "SCALE FACTOR",
1652            LoadGeneratorOptionName::TickInterval => "TICK INTERVAL",
1653            LoadGeneratorOptionName::AsOf => "AS OF",
1654            LoadGeneratorOptionName::UpTo => "UP TO",
1655            LoadGeneratorOptionName::MaxCardinality => "MAX CARDINALITY",
1656            LoadGeneratorOptionName::Keys => "KEYS",
1657            LoadGeneratorOptionName::SnapshotRounds => "SNAPSHOT ROUNDS",
1658            LoadGeneratorOptionName::TransactionalSnapshot => "TRANSACTIONAL SNAPSHOT",
1659            LoadGeneratorOptionName::ValueSize => "VALUE SIZE",
1660            LoadGeneratorOptionName::Seed => "SEED",
1661            LoadGeneratorOptionName::Partitions => "PARTITIONS",
1662            LoadGeneratorOptionName::BatchSize => "BATCH SIZE",
1663        })
1664    }
1665}
1666impl_display!(LoadGeneratorOptionName);
1667
1668impl WithOptionName for LoadGeneratorOptionName {
1669    /// # WARNING
1670    ///
1671    /// Whenever implementing this trait consider very carefully whether or not
1672    /// this value could contain sensitive user data. If you're uncertain, err
1673    /// on the conservative side and return `true`.
1674    fn redact_value(&self) -> bool {
1675        match self {
1676            LoadGeneratorOptionName::ScaleFactor
1677            | LoadGeneratorOptionName::TickInterval
1678            | LoadGeneratorOptionName::AsOf
1679            | LoadGeneratorOptionName::UpTo
1680            | LoadGeneratorOptionName::MaxCardinality
1681            | LoadGeneratorOptionName::Keys
1682            | LoadGeneratorOptionName::SnapshotRounds
1683            | LoadGeneratorOptionName::TransactionalSnapshot
1684            | LoadGeneratorOptionName::ValueSize
1685            | LoadGeneratorOptionName::Partitions
1686            | LoadGeneratorOptionName::BatchSize
1687            | LoadGeneratorOptionName::Seed => false,
1688        }
1689    }
1690}
1691
1692#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1693/// An option in a `CREATE CONNECTION...SSH`.
1694pub struct LoadGeneratorOption<T: AstInfo> {
1695    pub name: LoadGeneratorOptionName,
1696    pub value: Option<WithOptionValue<T>>,
1697}
1698impl_display_for_with_option!(LoadGeneratorOption);
1699impl_display_t!(LoadGeneratorOption);
1700
1701#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1702pub enum CreateSinkConnection<T: AstInfo> {
1703    Kafka {
1704        connection: T::ItemName,
1705        options: Vec<KafkaSinkConfigOption<T>>,
1706        key: Option<SinkKey>,
1707        headers: Option<Ident>,
1708    },
1709    Iceberg {
1710        catalog_connection: T::ItemName,
1711
1712        /// AWS creds for the storage layer.
1713        aws_connection: Option<T::ItemName>,
1714
1715        key: Option<SinkKey>,
1716        options: Vec<IcebergSinkConfigOption<T>>,
1717    },
1718}
1719
1720impl<T: AstInfo> AstDisplay for CreateSinkConnection<T> {
1721    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1722        match self {
1723            CreateSinkConnection::Kafka {
1724                connection,
1725                options,
1726                key,
1727                headers,
1728            } => {
1729                f.write_str("KAFKA CONNECTION ");
1730                f.write_node(connection);
1731                if !options.is_empty() {
1732                    f.write_str(" (");
1733                    f.write_node(&display::comma_separated(options));
1734                    f.write_str(")");
1735                }
1736                if let Some(key) = key.as_ref() {
1737                    f.write_str(" ");
1738                    f.write_node(key);
1739                }
1740                if let Some(headers) = headers {
1741                    f.write_str(" HEADERS ");
1742                    f.write_node(headers);
1743                }
1744            }
1745            CreateSinkConnection::Iceberg {
1746                catalog_connection,
1747                aws_connection,
1748                key,
1749                options,
1750            } => {
1751                f.write_str("ICEBERG CATALOG CONNECTION ");
1752                f.write_node(catalog_connection);
1753                if !options.is_empty() {
1754                    f.write_str(" (");
1755                    f.write_node(&display::comma_separated(options));
1756                    f.write_str(")");
1757                }
1758                if let Some(aws_connection) = aws_connection {
1759                    f.write_str(" USING AWS CONNECTION ");
1760                    f.write_node(aws_connection);
1761                }
1762                if let Some(key) = key.as_ref() {
1763                    f.write_str(" ");
1764                    f.write_node(key);
1765                }
1766            }
1767        }
1768    }
1769}
1770impl_display_t!(CreateSinkConnection);
1771
1772#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1773pub struct SinkKey {
1774    pub key_columns: Vec<Ident>,
1775    pub not_enforced: bool,
1776}
1777
1778impl AstDisplay for SinkKey {
1779    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1780        f.write_str("KEY (");
1781        f.write_node(&display::comma_separated(&self.key_columns));
1782        f.write_str(")");
1783        if self.not_enforced {
1784            f.write_str(" NOT ENFORCED");
1785        }
1786    }
1787}
1788
1789/// A table-level constraint, specified in a `CREATE TABLE` or an
1790/// `ALTER TABLE ADD <constraint>` statement.
1791#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1792pub enum TableConstraint<T: AstInfo> {
1793    /// `[ CONSTRAINT <name> ] { PRIMARY KEY | UNIQUE (NULLS NOT DISTINCT)? } (<columns>)`
1794    Unique {
1795        name: Option<Ident>,
1796        columns: Vec<Ident>,
1797        /// Whether this is a `PRIMARY KEY` or just a `UNIQUE` constraint
1798        is_primary: bool,
1799        // Where this constraint treats each NULL value as distinct; only available on `UNIQUE`
1800        // constraints.
1801        nulls_not_distinct: bool,
1802    },
1803    /// A referential integrity constraint (`[ CONSTRAINT <name> ] FOREIGN KEY (<columns>)
1804    /// REFERENCES <foreign_table> (<referred_columns>)`)
1805    ForeignKey {
1806        name: Option<Ident>,
1807        columns: Vec<Ident>,
1808        foreign_table: T::ItemName,
1809        referred_columns: Vec<Ident>,
1810    },
1811    /// `[ CONSTRAINT <name> ] CHECK (<expr>)`
1812    Check {
1813        name: Option<Ident>,
1814        expr: Box<Expr<T>>,
1815    },
1816}
1817
1818impl<T: AstInfo> AstDisplay for TableConstraint<T> {
1819    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1820        match self {
1821            TableConstraint::Unique {
1822                name,
1823                columns,
1824                is_primary,
1825                nulls_not_distinct,
1826            } => {
1827                f.write_node(&display_constraint_name(name));
1828                if *is_primary {
1829                    f.write_str("PRIMARY KEY ");
1830                } else {
1831                    f.write_str("UNIQUE ");
1832                    if *nulls_not_distinct {
1833                        f.write_str("NULLS NOT DISTINCT ");
1834                    }
1835                }
1836                f.write_str("(");
1837                f.write_node(&display::comma_separated(columns));
1838                f.write_str(")");
1839            }
1840            TableConstraint::ForeignKey {
1841                name,
1842                columns,
1843                foreign_table,
1844                referred_columns,
1845            } => {
1846                f.write_node(&display_constraint_name(name));
1847                f.write_str("FOREIGN KEY (");
1848                f.write_node(&display::comma_separated(columns));
1849                f.write_str(") REFERENCES ");
1850                f.write_node(foreign_table);
1851                f.write_str("(");
1852                f.write_node(&display::comma_separated(referred_columns));
1853                f.write_str(")");
1854            }
1855            TableConstraint::Check { name, expr } => {
1856                f.write_node(&display_constraint_name(name));
1857                f.write_str("CHECK (");
1858                f.write_node(&expr);
1859                f.write_str(")");
1860            }
1861        }
1862    }
1863}
1864impl_display_t!(TableConstraint);
1865
1866/// A key constraint, specified in a `CREATE SOURCE`.
1867#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1868pub enum KeyConstraint {
1869    // PRIMARY KEY (<columns>) NOT ENFORCED
1870    PrimaryKeyNotEnforced { columns: Vec<Ident> },
1871}
1872
1873impl AstDisplay for KeyConstraint {
1874    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1875        match self {
1876            KeyConstraint::PrimaryKeyNotEnforced { columns } => {
1877                f.write_str("PRIMARY KEY ");
1878                f.write_str("(");
1879                f.write_node(&display::comma_separated(columns));
1880                f.write_str(") ");
1881                f.write_str("NOT ENFORCED");
1882            }
1883        }
1884    }
1885}
1886impl_display!(KeyConstraint);
1887
1888#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1889pub enum CreateSourceOptionName {
1890    TimestampInterval,
1891    RetainHistory,
1892}
1893
1894impl AstDisplay for CreateSourceOptionName {
1895    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1896        f.write_str(match self {
1897            CreateSourceOptionName::TimestampInterval => "TIMESTAMP INTERVAL",
1898            CreateSourceOptionName::RetainHistory => "RETAIN HISTORY",
1899        })
1900    }
1901}
1902impl_display!(CreateSourceOptionName);
1903
1904impl WithOptionName for CreateSourceOptionName {
1905    /// # WARNING
1906    ///
1907    /// Whenever implementing this trait consider very carefully whether or not
1908    /// this value could contain sensitive user data. If you're uncertain, err
1909    /// on the conservative side and return `true`.
1910    fn redact_value(&self) -> bool {
1911        match self {
1912            CreateSourceOptionName::TimestampInterval | CreateSourceOptionName::RetainHistory => {
1913                false
1914            }
1915        }
1916    }
1917}
1918
1919#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1920/// An option in a `CREATE SOURCE...` statement.
1921pub struct CreateSourceOption<T: AstInfo> {
1922    pub name: CreateSourceOptionName,
1923    pub value: Option<WithOptionValue<T>>,
1924}
1925impl_display_for_with_option!(CreateSourceOption);
1926impl_display_t!(CreateSourceOption);
1927
1928/// SQL column definition
1929#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1930pub struct ColumnDef<T: AstInfo> {
1931    pub name: Ident,
1932    pub data_type: T::DataType,
1933    pub collation: Option<UnresolvedItemName>,
1934    pub options: Vec<ColumnOptionDef<T>>,
1935}
1936
1937impl<T: AstInfo> AstDisplay for ColumnDef<T> {
1938    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1939        f.write_node(&self.name);
1940        f.write_str(" ");
1941        f.write_node(&self.data_type);
1942        if let Some(collation) = &self.collation {
1943            f.write_str(" COLLATE ");
1944            f.write_node(collation);
1945        }
1946        for option in &self.options {
1947            f.write_str(" ");
1948            f.write_node(option);
1949        }
1950    }
1951}
1952impl_display_t!(ColumnDef);
1953
1954/// An optionally-named `ColumnOption`: `[ CONSTRAINT <name> ] <column-option>`.
1955///
1956/// Note that implementations are substantially more permissive than the ANSI
1957/// specification on what order column options can be presented in, and whether
1958/// they are allowed to be named. The specification distinguishes between
1959/// constraints (NOT NULL, UNIQUE, PRIMARY KEY, and CHECK), which can be named
1960/// and can appear in any order, and other options (DEFAULT, GENERATED), which
1961/// cannot be named and must appear in a fixed order. PostgreSQL, however,
1962/// allows preceding any option with `CONSTRAINT <name>`, even those that are
1963/// not really constraints, like NULL and DEFAULT. MSSQL is less permissive,
1964/// allowing DEFAULT, UNIQUE, PRIMARY KEY and CHECK to be named, but not NULL or
1965/// NOT NULL constraints (the last of which is in violation of the spec).
1966///
1967/// For maximum flexibility, we don't distinguish between constraint and
1968/// non-constraint options, lumping them all together under the umbrella of
1969/// "column options," and we allow any column option to be named.
1970#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1971pub struct ColumnOptionDef<T: AstInfo> {
1972    pub name: Option<Ident>,
1973    pub option: ColumnOption<T>,
1974}
1975
1976impl<T: AstInfo> AstDisplay for ColumnOptionDef<T> {
1977    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1978        f.write_node(&display_constraint_name(&self.name));
1979        f.write_node(&self.option);
1980    }
1981}
1982impl_display_t!(ColumnOptionDef);
1983
1984/// `ColumnOption`s are modifiers that follow a column definition in a `CREATE
1985/// TABLE` statement.
1986#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1987pub enum ColumnOption<T: AstInfo> {
1988    /// `NULL`
1989    Null,
1990    /// `NOT NULL`
1991    NotNull,
1992    /// `DEFAULT <restricted-expr>`
1993    Default(Expr<T>),
1994    /// `{ PRIMARY KEY | UNIQUE }`
1995    Unique { is_primary: bool },
1996    /// A referential integrity constraint (`[FOREIGN KEY REFERENCES
1997    /// <foreign_table> (<referred_columns>)`).
1998    ForeignKey {
1999        foreign_table: UnresolvedItemName,
2000        referred_columns: Vec<Ident>,
2001    },
2002    /// `CHECK (<expr>)`
2003    Check(Expr<T>),
2004    /// `VERSION <action> <version>`
2005    Versioned {
2006        action: ColumnVersioned,
2007        version: Version,
2008    },
2009}
2010
2011impl<T: AstInfo> AstDisplay for ColumnOption<T> {
2012    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
2013        use ColumnOption::*;
2014        match self {
2015            Null => f.write_str("NULL"),
2016            NotNull => f.write_str("NOT NULL"),
2017            Default(expr) => {
2018                f.write_str("DEFAULT ");
2019                f.write_node(expr);
2020            }
2021            Unique { is_primary } => {
2022                if *is_primary {
2023                    f.write_str("PRIMARY KEY");
2024                } else {
2025                    f.write_str("UNIQUE");
2026                }
2027            }
2028            ForeignKey {
2029                foreign_table,
2030                referred_columns,
2031            } => {
2032                f.write_str("REFERENCES ");
2033                f.write_node(foreign_table);
2034                f.write_str(" (");
2035                f.write_node(&display::comma_separated(referred_columns));
2036                f.write_str(")");
2037            }
2038            Check(expr) => {
2039                f.write_str("CHECK (");
2040                f.write_node(expr);
2041                f.write_str(")");
2042            }
2043            Versioned { action, version } => {
2044                f.write_str("VERSION ");
2045                f.write_node(action);
2046                f.write_str(" ");
2047                f.write_node(version);
2048            }
2049        }
2050    }
2051}
2052impl_display_t!(ColumnOption);
2053
2054#[derive(Debug, Clone, PartialEq, Eq, Hash)]
2055pub enum ColumnVersioned {
2056    Added,
2057}
2058
2059impl AstDisplay for ColumnVersioned {
2060    fn fmt<W>(&self, f: &mut AstFormatter<W>)
2061    where
2062        W: fmt::Write,
2063    {
2064        match self {
2065            // TODO(alter_table): Support dropped columns.
2066            ColumnVersioned::Added => f.write_str("ADDED"),
2067        }
2068    }
2069}
2070impl_display!(ColumnVersioned);
2071
2072fn display_constraint_name<'a>(name: &'a Option<Ident>) -> impl AstDisplay + 'a {
2073    struct ConstraintName<'a>(&'a Option<Ident>);
2074    impl<'a> AstDisplay for ConstraintName<'a> {
2075        fn fmt<W>(&self, f: &mut AstFormatter<W>)
2076        where
2077            W: fmt::Write,
2078        {
2079            if let Some(name) = self.0 {
2080                f.write_str("CONSTRAINT ");
2081                f.write_node(name);
2082                f.write_str(" ");
2083            }
2084        }
2085    }
2086    ConstraintName(name)
2087}
mz_sql_parser/ast/defs/ddl.rs

mz_sql_parser/ast/defs/
ddl.rs
