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    /// Whether to create the replication slot with `FAILOVER`
1333    Failover,
1334    /// Columns whose types you want to unconditionally format as text
1335    /// NOTE(roshan): This value is kept around to allow round-tripping a
1336    /// `CREATE SOURCE` statement while we still allow creating implicit
1337    /// subsources from `CREATE SOURCE`, but will be removed once
1338    /// fully deprecating that feature and forcing users to use explicit
1339    /// `CREATE TABLE .. FROM SOURCE` statements
1340    TextColumns,
1341    /// Columns you want to exclude
1342    /// NOTE: This value is kept around to allow round-tripping a
1343    /// `CREATE SOURCE` statement while we still allow creating implicit
1344    /// subsources from `CREATE SOURCE`, but will be removed once
1345    /// fully deprecating that feature and forcing users to use explicit
1346    /// `CREATE TABLE .. FROM SOURCE` statements
1347    ExcludeColumns,
1348}
1349
1350impl AstDisplay for PgConfigOptionName {
1351    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1352        f.write_str(match self {
1353            PgConfigOptionName::Details => "DETAILS",
1354            PgConfigOptionName::Publication => "PUBLICATION",
1355            PgConfigOptionName::Failover => "FAILOVER",
1356            PgConfigOptionName::TextColumns => "TEXT COLUMNS",
1357            PgConfigOptionName::ExcludeColumns => "EXCLUDE COLUMNS",
1358        })
1359    }
1360}
1361impl_display!(PgConfigOptionName);
1362
1363impl WithOptionName for PgConfigOptionName {
1364    /// # WARNING
1365    ///
1366    /// Whenever implementing this trait consider very carefully whether or not
1367    /// this value could contain sensitive user data. If you're uncertain, err
1368    /// on the conservative side and return `true`.
1369    fn redact_value(&self) -> bool {
1370        match self {
1371            PgConfigOptionName::Details
1372            | PgConfigOptionName::Publication
1373            | PgConfigOptionName::Failover
1374            | PgConfigOptionName::TextColumns
1375            | PgConfigOptionName::ExcludeColumns => false,
1376        }
1377    }
1378}
1379
1380#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1381/// An option in a `{FROM|INTO} CONNECTION ...` statement.
1382pub struct PgConfigOption<T: AstInfo> {
1383    pub name: PgConfigOptionName,
1384    pub value: Option<WithOptionValue<T>>,
1385}
1386impl_display_for_with_option!(PgConfigOption);
1387impl_display_t!(PgConfigOption);
1388
1389#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1390pub enum MySqlConfigOptionName {
1391    /// Hex encoded string of binary serialization of
1392    /// `mz_storage_types::sources::mysql::MySqlSourceDetails`
1393    Details,
1394    /// Columns whose types you want to unconditionally format as text
1395    /// NOTE(roshan): This value is kept around to allow round-tripping a
1396    /// `CREATE SOURCE` statement while we still allow creating implicit
1397    /// subsources from `CREATE SOURCE`, but will be removed once
1398    /// fully deprecating that feature and forcing users to use explicit
1399    /// `CREATE TABLE .. FROM SOURCE` statements
1400    TextColumns,
1401    /// Columns you want to exclude
1402    /// NOTE(roshan): This value is kept around to allow round-tripping a
1403    /// `CREATE SOURCE` statement while we still allow creating implicit
1404    /// subsources from `CREATE SOURCE`, but will be removed once
1405    /// fully deprecating that feature and forcing users to use explicit
1406    /// `CREATE TABLE .. FROM SOURCE` statements
1407    ExcludeColumns,
1408}
1409
1410impl AstDisplay for MySqlConfigOptionName {
1411    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1412        f.write_str(match self {
1413            MySqlConfigOptionName::Details => "DETAILS",
1414            MySqlConfigOptionName::TextColumns => "TEXT COLUMNS",
1415            MySqlConfigOptionName::ExcludeColumns => "EXCLUDE COLUMNS",
1416        })
1417    }
1418}
1419impl_display!(MySqlConfigOptionName);
1420
1421impl WithOptionName for MySqlConfigOptionName {
1422    /// # WARNING
1423    ///
1424    /// Whenever implementing this trait consider very carefully whether or not
1425    /// this value could contain sensitive user data. If you're uncertain, err
1426    /// on the conservative side and return `true`.
1427    fn redact_value(&self) -> bool {
1428        match self {
1429            MySqlConfigOptionName::Details
1430            | MySqlConfigOptionName::TextColumns
1431            | MySqlConfigOptionName::ExcludeColumns => false,
1432        }
1433    }
1434}
1435
1436#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1437/// An option in a `{FROM|INTO} CONNECTION ...` statement.
1438pub struct MySqlConfigOption<T: AstInfo> {
1439    pub name: MySqlConfigOptionName,
1440    pub value: Option<WithOptionValue<T>>,
1441}
1442impl_display_for_with_option!(MySqlConfigOption);
1443impl_display_t!(MySqlConfigOption);
1444
1445#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1446pub enum SqlServerConfigOptionName {
1447    /// Hex encoded string of binary serialization of
1448    /// `mz_storage_types::sources::sql_server::SqlServerSourceDetails`.
1449    Details,
1450    /// Columns whose types you want to unconditionally format as text.
1451    ///
1452    /// NOTE(roshan): This value is kept around to allow round-tripping a
1453    /// `CREATE SOURCE` statement while we still allow creating implicit
1454    /// subsources from `CREATE SOURCE`, but will be removed once
1455    /// fully deprecating that feature and forcing users to use explicit
1456    /// `CREATE TABLE .. FROM SOURCE` statements
1457    TextColumns,
1458    /// Columns you want to exclude.
1459    ///
1460    /// NOTE(roshan): This value is kept around to allow round-tripping a
1461    /// `CREATE SOURCE` statement while we still allow creating implicit
1462    /// subsources from `CREATE SOURCE`, but will be removed once
1463    /// fully deprecating that feature and forcing users to use explicit
1464    /// `CREATE TABLE .. FROM SOURCE` statements
1465    ExcludeColumns,
1466}
1467
1468impl AstDisplay for SqlServerConfigOptionName {
1469    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1470        f.write_str(match self {
1471            SqlServerConfigOptionName::Details => "DETAILS",
1472            SqlServerConfigOptionName::TextColumns => "TEXT COLUMNS",
1473            SqlServerConfigOptionName::ExcludeColumns => "EXCLUDE COLUMNS",
1474        })
1475    }
1476}
1477impl_display!(SqlServerConfigOptionName);
1478
1479impl WithOptionName for SqlServerConfigOptionName {
1480    /// # WARNING
1481    ///
1482    /// Whenever implementing this trait consider very carefully whether or not
1483    /// this value could contain sensitive user data. If you're uncertain, err
1484    /// on the conservative side and return `true`.
1485    fn redact_value(&self) -> bool {
1486        match self {
1487            SqlServerConfigOptionName::Details
1488            | SqlServerConfigOptionName::TextColumns
1489            | SqlServerConfigOptionName::ExcludeColumns => false,
1490        }
1491    }
1492}
1493
1494#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1495/// An option in a `{FROM|INTO} CONNECTION ...` statement.
1496pub struct SqlServerConfigOption<T: AstInfo> {
1497    pub name: SqlServerConfigOptionName,
1498    pub value: Option<WithOptionValue<T>>,
1499}
1500impl_display_for_with_option!(SqlServerConfigOption);
1501impl_display_t!(SqlServerConfigOption);
1502
1503#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1504pub enum CreateSourceConnection<T: AstInfo> {
1505    Kafka {
1506        connection: T::ItemName,
1507        options: Vec<KafkaSourceConfigOption<T>>,
1508    },
1509    Postgres {
1510        connection: T::ItemName,
1511        options: Vec<PgConfigOption<T>>,
1512    },
1513    SqlServer {
1514        connection: T::ItemName,
1515        options: Vec<SqlServerConfigOption<T>>,
1516    },
1517    MySql {
1518        connection: T::ItemName,
1519        options: Vec<MySqlConfigOption<T>>,
1520    },
1521    LoadGenerator {
1522        generator: LoadGenerator,
1523        options: Vec<LoadGeneratorOption<T>>,
1524    },
1525}
1526
1527impl<T: AstInfo> AstDisplay for CreateSourceConnection<T> {
1528    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1529        match self {
1530            CreateSourceConnection::Kafka {
1531                connection,
1532                options,
1533            } => {
1534                f.write_str("KAFKA CONNECTION ");
1535                f.write_node(connection);
1536                if !options.is_empty() {
1537                    f.write_str(" (");
1538                    f.write_node(&display::comma_separated(options));
1539                    f.write_str(")");
1540                }
1541            }
1542            CreateSourceConnection::Postgres {
1543                connection,
1544                options,
1545            } => {
1546                f.write_str("POSTGRES CONNECTION ");
1547                f.write_node(connection);
1548                if !options.is_empty() {
1549                    f.write_str(" (");
1550                    f.write_node(&display::comma_separated(options));
1551                    f.write_str(")");
1552                }
1553            }
1554            CreateSourceConnection::SqlServer {
1555                connection,
1556                options,
1557            } => {
1558                f.write_str("SQL SERVER CONNECTION ");
1559                f.write_node(connection);
1560                if !options.is_empty() {
1561                    f.write_str(" (");
1562                    f.write_node(&display::comma_separated(options));
1563                    f.write_str(")");
1564                }
1565            }
1566            CreateSourceConnection::MySql {
1567                connection,
1568                options,
1569            } => {
1570                f.write_str("MYSQL CONNECTION ");
1571                f.write_node(connection);
1572                if !options.is_empty() {
1573                    f.write_str(" (");
1574                    f.write_node(&display::comma_separated(options));
1575                    f.write_str(")");
1576                }
1577            }
1578            CreateSourceConnection::LoadGenerator { generator, options } => {
1579                f.write_str("LOAD GENERATOR ");
1580                f.write_node(generator);
1581                if !options.is_empty() {
1582                    f.write_str(" (");
1583                    f.write_node(&display::comma_separated(options));
1584                    f.write_str(")");
1585                }
1586            }
1587        }
1588    }
1589}
1590impl_display_t!(CreateSourceConnection);
1591
1592#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1593pub enum LoadGenerator {
1594    Clock,
1595    Counter,
1596    Marketing,
1597    Auction,
1598    Datums,
1599    Tpch,
1600    KeyValue,
1601}
1602
1603impl AstDisplay for LoadGenerator {
1604    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1605        match self {
1606            Self::Counter => f.write_str("COUNTER"),
1607            Self::Clock => f.write_str("CLOCK"),
1608            Self::Marketing => f.write_str("MARKETING"),
1609            Self::Auction => f.write_str("AUCTION"),
1610            Self::Datums => f.write_str("DATUMS"),
1611            Self::Tpch => f.write_str("TPCH"),
1612            Self::KeyValue => f.write_str("KEY VALUE"),
1613        }
1614    }
1615}
1616impl_display!(LoadGenerator);
1617
1618impl LoadGenerator {
1619    /// Corresponds with the same mapping on the `LoadGenerator` enum defined in
1620    /// src/storage-types/src/sources/load_generator.rs, but re-defined here for
1621    /// cases where we only have the AST representation. This can be removed once
1622    /// the `ast_rewrite_sources_to_tables` migration is removed.
1623    pub fn schema_name(&self) -> &'static str {
1624        match self {
1625            LoadGenerator::Counter => "counter",
1626            LoadGenerator::Clock => "clock",
1627            LoadGenerator::Marketing => "marketing",
1628            LoadGenerator::Auction => "auction",
1629            LoadGenerator::Datums => "datums",
1630            LoadGenerator::Tpch => "tpch",
1631            LoadGenerator::KeyValue => "key_value",
1632        }
1633    }
1634}
1635
1636#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1637pub enum LoadGeneratorOptionName {
1638    ScaleFactor,
1639    TickInterval,
1640    AsOf,
1641    UpTo,
1642    MaxCardinality,
1643    Keys,
1644    SnapshotRounds,
1645    TransactionalSnapshot,
1646    ValueSize,
1647    Seed,
1648    Partitions,
1649    BatchSize,
1650}
1651
1652impl AstDisplay for LoadGeneratorOptionName {
1653    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1654        f.write_str(match self {
1655            LoadGeneratorOptionName::ScaleFactor => "SCALE FACTOR",
1656            LoadGeneratorOptionName::TickInterval => "TICK INTERVAL",
1657            LoadGeneratorOptionName::AsOf => "AS OF",
1658            LoadGeneratorOptionName::UpTo => "UP TO",
1659            LoadGeneratorOptionName::MaxCardinality => "MAX CARDINALITY",
1660            LoadGeneratorOptionName::Keys => "KEYS",
1661            LoadGeneratorOptionName::SnapshotRounds => "SNAPSHOT ROUNDS",
1662            LoadGeneratorOptionName::TransactionalSnapshot => "TRANSACTIONAL SNAPSHOT",
1663            LoadGeneratorOptionName::ValueSize => "VALUE SIZE",
1664            LoadGeneratorOptionName::Seed => "SEED",
1665            LoadGeneratorOptionName::Partitions => "PARTITIONS",
1666            LoadGeneratorOptionName::BatchSize => "BATCH SIZE",
1667        })
1668    }
1669}
1670impl_display!(LoadGeneratorOptionName);
1671
1672impl WithOptionName for LoadGeneratorOptionName {
1673    /// # WARNING
1674    ///
1675    /// Whenever implementing this trait consider very carefully whether or not
1676    /// this value could contain sensitive user data. If you're uncertain, err
1677    /// on the conservative side and return `true`.
1678    fn redact_value(&self) -> bool {
1679        match self {
1680            LoadGeneratorOptionName::ScaleFactor
1681            | LoadGeneratorOptionName::TickInterval
1682            | LoadGeneratorOptionName::AsOf
1683            | LoadGeneratorOptionName::UpTo
1684            | LoadGeneratorOptionName::MaxCardinality
1685            | LoadGeneratorOptionName::Keys
1686            | LoadGeneratorOptionName::SnapshotRounds
1687            | LoadGeneratorOptionName::TransactionalSnapshot
1688            | LoadGeneratorOptionName::ValueSize
1689            | LoadGeneratorOptionName::Partitions
1690            | LoadGeneratorOptionName::BatchSize
1691            | LoadGeneratorOptionName::Seed => false,
1692        }
1693    }
1694}
1695
1696#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1697/// An option in a `CREATE CONNECTION...SSH`.
1698pub struct LoadGeneratorOption<T: AstInfo> {
1699    pub name: LoadGeneratorOptionName,
1700    pub value: Option<WithOptionValue<T>>,
1701}
1702impl_display_for_with_option!(LoadGeneratorOption);
1703impl_display_t!(LoadGeneratorOption);
1704
1705#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1706pub enum CreateSinkConnection<T: AstInfo> {
1707    Kafka {
1708        connection: T::ItemName,
1709        options: Vec<KafkaSinkConfigOption<T>>,
1710        key: Option<SinkKey>,
1711        headers: Option<Ident>,
1712    },
1713    Iceberg {
1714        catalog_connection: T::ItemName,
1715
1716        /// AWS creds for the storage layer.
1717        aws_connection: Option<T::ItemName>,
1718
1719        key: Option<SinkKey>,
1720        options: Vec<IcebergSinkConfigOption<T>>,
1721    },
1722}
1723
1724impl<T: AstInfo> AstDisplay for CreateSinkConnection<T> {
1725    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1726        match self {
1727            CreateSinkConnection::Kafka {
1728                connection,
1729                options,
1730                key,
1731                headers,
1732            } => {
1733                f.write_str("KAFKA CONNECTION ");
1734                f.write_node(connection);
1735                if !options.is_empty() {
1736                    f.write_str(" (");
1737                    f.write_node(&display::comma_separated(options));
1738                    f.write_str(")");
1739                }
1740                if let Some(key) = key.as_ref() {
1741                    f.write_str(" ");
1742                    f.write_node(key);
1743                }
1744                if let Some(headers) = headers {
1745                    f.write_str(" HEADERS ");
1746                    f.write_node(headers);
1747                }
1748            }
1749            CreateSinkConnection::Iceberg {
1750                catalog_connection,
1751                aws_connection,
1752                key,
1753                options,
1754            } => {
1755                f.write_str("ICEBERG CATALOG CONNECTION ");
1756                f.write_node(catalog_connection);
1757                if !options.is_empty() {
1758                    f.write_str(" (");
1759                    f.write_node(&display::comma_separated(options));
1760                    f.write_str(")");
1761                }
1762                if let Some(aws_connection) = aws_connection {
1763                    f.write_str(" USING AWS CONNECTION ");
1764                    f.write_node(aws_connection);
1765                }
1766                if let Some(key) = key.as_ref() {
1767                    f.write_str(" ");
1768                    f.write_node(key);
1769                }
1770            }
1771        }
1772    }
1773}
1774impl_display_t!(CreateSinkConnection);
1775
1776#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1777pub struct SinkKey {
1778    pub key_columns: Vec<Ident>,
1779    pub not_enforced: bool,
1780}
1781
1782impl AstDisplay for SinkKey {
1783    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1784        f.write_str("KEY (");
1785        f.write_node(&display::comma_separated(&self.key_columns));
1786        f.write_str(")");
1787        if self.not_enforced {
1788            f.write_str(" NOT ENFORCED");
1789        }
1790    }
1791}
1792
1793/// A table-level constraint, specified in a `CREATE TABLE` or an
1794/// `ALTER TABLE ADD <constraint>` statement.
1795#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1796pub enum TableConstraint<T: AstInfo> {
1797    /// `[ CONSTRAINT <name> ] { PRIMARY KEY | UNIQUE (NULLS NOT DISTINCT)? } (<columns>)`
1798    Unique {
1799        name: Option<Ident>,
1800        columns: Vec<Ident>,
1801        /// Whether this is a `PRIMARY KEY` or just a `UNIQUE` constraint
1802        is_primary: bool,
1803        // Where this constraint treats each NULL value as distinct; only available on `UNIQUE`
1804        // constraints.
1805        nulls_not_distinct: bool,
1806    },
1807    /// A referential integrity constraint (`[ CONSTRAINT <name> ] FOREIGN KEY (<columns>)
1808    /// REFERENCES <foreign_table> (<referred_columns>)`)
1809    ForeignKey {
1810        name: Option<Ident>,
1811        columns: Vec<Ident>,
1812        foreign_table: T::ItemName,
1813        referred_columns: Vec<Ident>,
1814    },
1815    /// `[ CONSTRAINT <name> ] CHECK (<expr>)`
1816    Check {
1817        name: Option<Ident>,
1818        expr: Box<Expr<T>>,
1819    },
1820}
1821
1822impl<T: AstInfo> AstDisplay for TableConstraint<T> {
1823    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1824        match self {
1825            TableConstraint::Unique {
1826                name,
1827                columns,
1828                is_primary,
1829                nulls_not_distinct,
1830            } => {
1831                f.write_node(&display_constraint_name(name));
1832                if *is_primary {
1833                    f.write_str("PRIMARY KEY ");
1834                } else {
1835                    f.write_str("UNIQUE ");
1836                    if *nulls_not_distinct {
1837                        f.write_str("NULLS NOT DISTINCT ");
1838                    }
1839                }
1840                f.write_str("(");
1841                f.write_node(&display::comma_separated(columns));
1842                f.write_str(")");
1843            }
1844            TableConstraint::ForeignKey {
1845                name,
1846                columns,
1847                foreign_table,
1848                referred_columns,
1849            } => {
1850                f.write_node(&display_constraint_name(name));
1851                f.write_str("FOREIGN KEY (");
1852                f.write_node(&display::comma_separated(columns));
1853                f.write_str(") REFERENCES ");
1854                f.write_node(foreign_table);
1855                f.write_str("(");
1856                f.write_node(&display::comma_separated(referred_columns));
1857                f.write_str(")");
1858            }
1859            TableConstraint::Check { name, expr } => {
1860                f.write_node(&display_constraint_name(name));
1861                f.write_str("CHECK (");
1862                f.write_node(&expr);
1863                f.write_str(")");
1864            }
1865        }
1866    }
1867}
1868impl_display_t!(TableConstraint);
1869
1870/// A key constraint, specified in a `CREATE SOURCE`.
1871#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1872pub enum KeyConstraint {
1873    // PRIMARY KEY (<columns>) NOT ENFORCED
1874    PrimaryKeyNotEnforced { columns: Vec<Ident> },
1875}
1876
1877impl AstDisplay for KeyConstraint {
1878    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1879        match self {
1880            KeyConstraint::PrimaryKeyNotEnforced { columns } => {
1881                f.write_str("PRIMARY KEY ");
1882                f.write_str("(");
1883                f.write_node(&display::comma_separated(columns));
1884                f.write_str(") ");
1885                f.write_str("NOT ENFORCED");
1886            }
1887        }
1888    }
1889}
1890impl_display!(KeyConstraint);
1891
1892#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1893pub enum CreateSourceOptionName {
1894    TimestampInterval,
1895    RetainHistory,
1896}
1897
1898impl AstDisplay for CreateSourceOptionName {
1899    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1900        f.write_str(match self {
1901            CreateSourceOptionName::TimestampInterval => "TIMESTAMP INTERVAL",
1902            CreateSourceOptionName::RetainHistory => "RETAIN HISTORY",
1903        })
1904    }
1905}
1906impl_display!(CreateSourceOptionName);
1907
1908impl WithOptionName for CreateSourceOptionName {
1909    /// # WARNING
1910    ///
1911    /// Whenever implementing this trait consider very carefully whether or not
1912    /// this value could contain sensitive user data. If you're uncertain, err
1913    /// on the conservative side and return `true`.
1914    fn redact_value(&self) -> bool {
1915        match self {
1916            CreateSourceOptionName::TimestampInterval | CreateSourceOptionName::RetainHistory => {
1917                false
1918            }
1919        }
1920    }
1921}
1922
1923#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
1924/// An option in a `CREATE SOURCE...` statement.
1925pub struct CreateSourceOption<T: AstInfo> {
1926    pub name: CreateSourceOptionName,
1927    pub value: Option<WithOptionValue<T>>,
1928}
1929impl_display_for_with_option!(CreateSourceOption);
1930impl_display_t!(CreateSourceOption);
1931
1932/// SQL column definition
1933#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1934pub struct ColumnDef<T: AstInfo> {
1935    pub name: Ident,
1936    pub data_type: T::DataType,
1937    pub collation: Option<UnresolvedItemName>,
1938    pub options: Vec<ColumnOptionDef<T>>,
1939}
1940
1941impl<T: AstInfo> AstDisplay for ColumnDef<T> {
1942    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1943        f.write_node(&self.name);
1944        f.write_str(" ");
1945        f.write_node(&self.data_type);
1946        if let Some(collation) = &self.collation {
1947            f.write_str(" COLLATE ");
1948            f.write_node(collation);
1949        }
1950        for option in &self.options {
1951            f.write_str(" ");
1952            f.write_node(option);
1953        }
1954    }
1955}
1956impl_display_t!(ColumnDef);
1957
1958/// An optionally-named `ColumnOption`: `[ CONSTRAINT <name> ] <column-option>`.
1959///
1960/// Note that implementations are substantially more permissive than the ANSI
1961/// specification on what order column options can be presented in, and whether
1962/// they are allowed to be named. The specification distinguishes between
1963/// constraints (NOT NULL, UNIQUE, PRIMARY KEY, and CHECK), which can be named
1964/// and can appear in any order, and other options (DEFAULT, GENERATED), which
1965/// cannot be named and must appear in a fixed order. PostgreSQL, however,
1966/// allows preceding any option with `CONSTRAINT <name>`, even those that are
1967/// not really constraints, like NULL and DEFAULT. MSSQL is less permissive,
1968/// allowing DEFAULT, UNIQUE, PRIMARY KEY and CHECK to be named, but not NULL or
1969/// NOT NULL constraints (the last of which is in violation of the spec).
1970///
1971/// For maximum flexibility, we don't distinguish between constraint and
1972/// non-constraint options, lumping them all together under the umbrella of
1973/// "column options," and we allow any column option to be named.
1974#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1975pub struct ColumnOptionDef<T: AstInfo> {
1976    pub name: Option<Ident>,
1977    pub option: ColumnOption<T>,
1978}
1979
1980impl<T: AstInfo> AstDisplay for ColumnOptionDef<T> {
1981    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
1982        f.write_node(&display_constraint_name(&self.name));
1983        f.write_node(&self.option);
1984    }
1985}
1986impl_display_t!(ColumnOptionDef);
1987
1988/// `ColumnOption`s are modifiers that follow a column definition in a `CREATE
1989/// TABLE` statement.
1990#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1991pub enum ColumnOption<T: AstInfo> {
1992    /// `NULL`
1993    Null,
1994    /// `NOT NULL`
1995    NotNull,
1996    /// `DEFAULT <restricted-expr>`
1997    Default(Expr<T>),
1998    /// `{ PRIMARY KEY | UNIQUE }`
1999    Unique { is_primary: bool },
2000    /// A referential integrity constraint (`[FOREIGN KEY REFERENCES
2001    /// <foreign_table> (<referred_columns>)`).
2002    ForeignKey {
2003        foreign_table: UnresolvedItemName,
2004        referred_columns: Vec<Ident>,
2005    },
2006    /// `CHECK (<expr>)`
2007    Check(Expr<T>),
2008    /// `VERSION <action> <version>`
2009    Versioned {
2010        action: ColumnVersioned,
2011        version: Version,
2012    },
2013}
2014
2015impl<T: AstInfo> AstDisplay for ColumnOption<T> {
2016    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
2017        use ColumnOption::*;
2018        match self {
2019            Null => f.write_str("NULL"),
2020            NotNull => f.write_str("NOT NULL"),
2021            Default(expr) => {
2022                f.write_str("DEFAULT ");
2023                f.write_node(expr);
2024            }
2025            Unique { is_primary } => {
2026                if *is_primary {
2027                    f.write_str("PRIMARY KEY");
2028                } else {
2029                    f.write_str("UNIQUE");
2030                }
2031            }
2032            ForeignKey {
2033                foreign_table,
2034                referred_columns,
2035            } => {
2036                f.write_str("REFERENCES ");
2037                f.write_node(foreign_table);
2038                f.write_str(" (");
2039                f.write_node(&display::comma_separated(referred_columns));
2040                f.write_str(")");
2041            }
2042            Check(expr) => {
2043                f.write_str("CHECK (");
2044                f.write_node(expr);
2045                f.write_str(")");
2046            }
2047            Versioned { action, version } => {
2048                f.write_str("VERSION ");
2049                f.write_node(action);
2050                f.write_str(" ");
2051                f.write_node(version);
2052            }
2053        }
2054    }
2055}
2056impl_display_t!(ColumnOption);
2057
2058#[derive(Debug, Clone, PartialEq, Eq, Hash)]
2059pub enum ColumnVersioned {
2060    Added,
2061}
2062
2063impl AstDisplay for ColumnVersioned {
2064    fn fmt<W>(&self, f: &mut AstFormatter<W>)
2065    where
2066        W: fmt::Write,
2067    {
2068        match self {
2069            // TODO(alter_table): Support dropped columns.
2070            ColumnVersioned::Added => f.write_str("ADDED"),
2071        }
2072    }
2073}
2074impl_display!(ColumnVersioned);
2075
2076fn display_constraint_name<'a>(name: &'a Option<Ident>) -> impl AstDisplay + 'a {
2077    struct ConstraintName<'a>(&'a Option<Ident>);
2078    impl<'a> AstDisplay for ConstraintName<'a> {
2079        fn fmt<W>(&self, f: &mut AstFormatter<W>)
2080        where
2081            W: fmt::Write,
2082        {
2083            if let Some(name) = self.0 {
2084                f.write_str("CONSTRAINT ");
2085                f.write_node(name);
2086                f.write_str(" ");
2087            }
2088        }
2089    }
2090    ConstraintName(name)
2091}
mz_sql_parser/ast/defs/ddl.rs

mz_sql_parser/ast/defs/
ddl.rs
