mz_sql/pure/
error.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10use std::sync::Arc;
11
12use mz_ccsr::ListError;
13use mz_repr::adt::system::Oid;
14use mz_sql_parser::ast::display::AstDisplay;
15use mz_sql_parser::ast::{ExternalReferences, UnresolvedItemName};
16use mz_storage_types::connections::{
17    MySqlConnectionValidationError, PostgresConnectionValidationError,
18};
19use mz_storage_types::errors::{ContextCreationError, CsrConnectError};
20
21use crate::names::{FullItemName, PartialItemName};
22
23/// Logical errors detectable during purification for a POSTGRES SOURCE.
24#[derive(Debug, Clone, thiserror::Error)]
25pub enum PgSourcePurificationError {
26    #[error("CREATE SOURCE specifies DETAILS option")]
27    UserSpecifiedDetails,
28    #[error("{0} option is unnecessary when no tables are added")]
29    UnnecessaryOptionsWithoutReferences(String),
30    #[error("PUBLICATION {0} is empty")]
31    EmptyPublication(String),
32    #[error("database {database} missing referenced schemas")]
33    DatabaseMissingFilteredSchemas {
34        database: String,
35        schemas: Vec<String>,
36    },
37    #[error("missing TABLES specification")]
38    RequiresExternalReferences,
39    #[error("insufficient privileges")]
40    UserLacksUsageOnSchemas { schemas: Vec<String> },
41    #[error("insufficient privileges")]
42    UserLacksSelectOnTables { tables: Vec<String> },
43    #[error("one or more tables requires BYPASSRLS")]
44    BypassRLSRequired { tables: Vec<String> },
45    #[error("referenced items not tables with REPLICA IDENTITY FULL")]
46    NotTablesWReplicaIdentityFull { items: Vec<String> },
47    #[error("TEXT COLUMNS refers to table not currently being added")]
48    DanglingTextColumns { items: Vec<PartialItemName> },
49    #[error("EXCLUDE COLUMNS refers to table not currently being added")]
50    DanglingExcludeColumns { items: Vec<PartialItemName> },
51    #[error("EXCLUDE CONSTRAINTS refers to constraints that do not exist on table {table}")]
52    ConstraintsNotFound {
53        table: PartialItemName,
54        constraints: Vec<String>,
55    },
56    #[error("duplicated column name references: {0:?}")]
57    DuplicatedColumnNames(Vec<String>),
58    #[error("referenced tables use unsupported types")]
59    UnrecognizedTypes { cols: Vec<(String, Oid)> },
60    #[error("{0} is not a POSTGRES CONNECTION")]
61    NotPgConnection(FullItemName),
62    #[error("CONNECTION must specify PUBLICATION")]
63    ConnectionMissingPublication,
64    #[error("FAILOVER requires the upstream to be a PostgreSQL 17+ primary")]
65    FailoverUnsupported,
66    #[error(transparent)]
67    InvalidConnection(PostgresConnectionValidationError),
68}
69
70impl PgSourcePurificationError {
71    pub fn detail(&self) -> Option<String> {
72        match self {
73            Self::DanglingTextColumns { items } => Some(format!(
74                "the following tables are referenced but not added: {}",
75                itertools::join(items, ", ")
76            )),
77            Self::ConstraintsNotFound {
78                table: _,
79                constraints,
80            } => Some(format!(
81                "the following constraints were not found: {}",
82                constraints.join(", ")
83            )),
84            Self::DatabaseMissingFilteredSchemas {
85                database: _,
86                schemas,
87            } => Some(format!(
88                "missing schemas: {}",
89                itertools::join(schemas.iter(), ", ")
90            )),
91            Self::UserLacksUsageOnSchemas { schemas } => Some(format!(
92                "user lacks USAGE privileges for schemas {}",
93                schemas.join(", ")
94            )),
95            Self::UserLacksSelectOnTables { tables } => Some(format!(
96                "user lacks SELECT privileges for tables {}",
97                tables.join(", ")
98            )),
99            Self::BypassRLSRequired { tables } => Some(format!(
100                "user must have BYPASSRLS attribute to read tables {}",
101                tables.join(", "),
102            )),
103            Self::NotTablesWReplicaIdentityFull { items } => {
104                Some(format!("referenced items: {}", items.join(", ")))
105            }
106            Self::UnrecognizedTypes { cols } => Some(format!(
107                "the following columns contain unsupported types:\n{}",
108                itertools::join(
109                    cols.into_iter()
110                        .map(|(col, Oid(oid))| format!("{} (OID {})", col, oid)),
111                    "\n"
112                )
113            )),
114            Self::InvalidConnection(e) => e.detail(),
115            _ => None,
116        }
117    }
118
119    pub fn hint(&self) -> Option<String> {
120        match self {
121            Self::UserSpecifiedDetails => Some(
122                "If trying to use the output of SHOW CREATE SOURCE, remove the DETAILS option."
123                    .into(),
124            ),
125            Self::RequiresExternalReferences => {
126                Some("provide a FOR TABLES (..), FOR SCHEMAS (..), or FOR ALL TABLES clause".into())
127            }
128            Self::UnrecognizedTypes {
129                cols: _,
130            } => Some(
131                "Use the TEXT COLUMNS option naming the listed columns, and Materialize can ingest their values \
132                as text."
133                    .into(),
134            ),
135            Self::UnnecessaryOptionsWithoutReferences(option) => Some(format!(
136                "Remove the {} option, as no tables are being added.",
137                option
138            )),
139            Self::ConstraintsNotFound { .. } => Some(
140                "Constraint names are matched exactly, including case, against the upstream \
141                 PRIMARY KEY and UNIQUE constraint names."
142                    .into(),
143            ),
144            Self::BypassRLSRequired { .. } => Some("Add the BYPASSRLS attribute to the Materialize user".into()),
145            Self::InvalidConnection(e) => e.hint(),
146            _ => None,
147        }
148    }
149}
150
151/// Logical errors detectable during purification for a KAFKA SOURCE.
152#[derive(Debug, Clone, thiserror::Error)]
153pub enum KafkaSourcePurificationError {
154    #[error("{} is only valid for multi-output sources", .0.to_ast_string_simple())]
155    ReferencedSubsources(ExternalReferences),
156    #[error("KAFKA CONNECTION without TOPIC")]
157    ConnectionMissingTopic,
158    #[error("{0} is not a KAFKA CONNECTION")]
159    NotKafkaConnection(FullItemName),
160    #[error("failed to create and connect Kafka consumer")]
161    KafkaConsumerError(String),
162    #[error("Referenced kafka connection uses a different topic '{0}' than specified: '{1}'")]
163    WrongKafkaTopic(String, UnresolvedItemName),
164}
165
166impl KafkaSourcePurificationError {
167    pub fn detail(&self) -> Option<String> {
168        match self {
169            Self::KafkaConsumerError(e) => Some(e.clone()),
170            _ => None,
171        }
172    }
173
174    pub fn hint(&self) -> Option<String> {
175        None
176    }
177}
178
179/// Logical errors detectable during purification for a LOAD GENERATOR SOURCE.
180#[derive(Debug, Clone, thiserror::Error)]
181pub enum LoadGeneratorSourcePurificationError {
182    #[error("FOR ALL TABLES is only valid for multi-output sources")]
183    ForAllTables,
184    #[error("FOR SCHEMAS (..) unsupported")]
185    ForSchemas,
186    #[error("FOR TABLES (..) unsupported")]
187    ForTables,
188    #[error("multi-output sources require a FOR TABLES (..) or FOR ALL TABLES statement")]
189    MultiOutputRequiresForAllTables,
190    #[error("multi-output sources require an external reference")]
191    MultiOutputRequiresExternalReference,
192    #[error("Referenced load generator is different '{0}' than specified: '{1}'")]
193    WrongLoadGenerator(String, UnresolvedItemName),
194}
195
196impl LoadGeneratorSourcePurificationError {
197    pub fn detail(&self) -> Option<String> {
198        match self {
199            _ => None,
200        }
201    }
202
203    pub fn hint(&self) -> Option<String> {
204        match self {
205            _ => None,
206        }
207    }
208}
209
210/// Logical errors detectable during purification for a KAFKA SINK.
211#[derive(Debug, Clone, thiserror::Error)]
212pub enum KafkaSinkPurificationError {
213    #[error("{0} is not a KAFKA CONNECTION")]
214    NotKafkaConnection(FullItemName),
215    #[error("admin client errored")]
216    AdminClientError(Arc<ContextCreationError>),
217    #[error("zero brokers discovered in metadata request")]
218    ZeroBrokers,
219}
220
221impl KafkaSinkPurificationError {
222    pub fn detail(&self) -> Option<String> {
223        match self {
224            Self::AdminClientError(e) => Some(e.to_string_with_causes()),
225            _ => None,
226        }
227    }
228
229    pub fn hint(&self) -> Option<String> {
230        None
231    }
232}
233
234#[derive(Debug, Clone, thiserror::Error)]
235pub enum IcebergSinkPurificationError {
236    #[error("catalog connection errored")]
237    CatalogError(Arc<anyhow::Error>),
238    #[error("error loading aws sdk context")]
239    AwsSdkContextError(Arc<anyhow::Error>),
240    #[error("S3 Tables connection region mismatch")]
241    S3TablesRegionMismatch {
242        s3_tables_region: String,
243        environment_region: String,
244    },
245}
246
247impl IcebergSinkPurificationError {
248    pub fn detail(&self) -> Option<String> {
249        match self {
250            Self::CatalogError(e) => Some(e.to_string_with_causes()),
251            Self::AwsSdkContextError(e) => Some(e.to_string_with_causes()),
252            Self::S3TablesRegionMismatch {
253                s3_tables_region,
254                environment_region,
255            } => Some(format!(
256                "S3 Tables connection is configured for region '{}' but this Materialize environment is running in region '{}'",
257                s3_tables_region, environment_region
258            )),
259        }
260    }
261
262    pub fn hint(&self) -> Option<String> {
263        match self {
264            Self::S3TablesRegionMismatch {
265                environment_region, ..
266            } => Some(format!(
267                "Create a new AWS connection with REGION = '{}' to use with S3 Tables in this environment.",
268                environment_region
269            )),
270            _ => None,
271        }
272    }
273}
274
275use mz_ore::error::ErrorExt;
276
277/// Logical errors detectable during purification for Confluent Schema Registry.
278#[derive(Debug, Clone, thiserror::Error)]
279pub enum CsrPurificationError {
280    #[error("{0} is not a CONFLUENT SCHEMA REGISTRY CONNECTION")]
281    NotCsrConnection(FullItemName),
282    #[error("client errored")]
283    ClientError(Arc<CsrConnectError>),
284    #[error("list subjects failed")]
285    ListSubjectsError(Arc<ListError>),
286}
287
288impl CsrPurificationError {
289    pub fn detail(&self) -> Option<String> {
290        match self {
291            Self::ClientError(e) => Some(e.to_string_with_causes()),
292            Self::ListSubjectsError(e) => Some(e.to_string_with_causes()),
293            Self::NotCsrConnection(_) => None,
294        }
295    }
296
297    pub fn hint(&self) -> Option<String> {
298        None
299    }
300}
301
302/// Logical errors detectable during purification for AWS Glue Schema Registry.
303#[derive(Debug, Clone, thiserror::Error)]
304pub enum GluePurificationError {
305    #[error("{0} is not an AWS GLUE SCHEMA REGISTRY CONNECTION")]
306    NotGlueConnection(FullItemName),
307    #[error("SCHEMA NAME option is required")]
308    MissingSchemaName,
309    #[error("loading AWS SDK configuration failed")]
310    LoadSdkConfigError(Arc<anyhow::Error>),
311    #[error("Glue schema lookup failed (registry {registry:?}, schema {schema:?})")]
312    SchemaLookupError {
313        registry: String,
314        schema: String,
315        #[source]
316        cause: Arc<mz_aws_glue_schema_registry::GetSchemaVersionError>,
317    },
318    #[error("Glue schema {schema:?} in registry {registry:?} has no definition")]
319    EmptyDefinition { registry: String, schema: String },
320    #[error(
321        "Glue schema {schema:?} in registry {registry:?} uses unsupported data format {format}; only AVRO is supported"
322    )]
323    UnsupportedDataFormat {
324        registry: String,
325        schema: String,
326        format: String,
327    },
328}
329
330impl GluePurificationError {
331    pub fn detail(&self) -> Option<String> {
332        match self {
333            Self::LoadSdkConfigError(e) => Some(e.to_string_with_causes()),
334            Self::SchemaLookupError { cause, .. } => Some(cause.to_string_with_causes()),
335            Self::NotGlueConnection(_)
336            | Self::MissingSchemaName
337            | Self::EmptyDefinition { .. }
338            | Self::UnsupportedDataFormat { .. } => None,
339        }
340    }
341
342    pub fn hint(&self) -> Option<String> {
343        None
344    }
345}
346
347/// Logical errors detectable during purification for a MySQL SOURCE.
348#[derive(Debug, thiserror::Error)]
349pub enum MySqlSourcePurificationError {
350    #[error("User lacks required MySQL privileges")]
351    UserLacksPrivileges(Vec<(String, String)>),
352    #[error("CREATE SOURCE specifies DETAILS option")]
353    UserSpecifiedDetails,
354    #[error("{0} option is unnecessary when no tables are added")]
355    UnnecessaryOptionsWithoutReferences(String),
356    #[error("{0} is not a MYSQL CONNECTION")]
357    NotMySqlConnection(FullItemName),
358    #[error("referenced tables use unsupported types")]
359    UnrecognizedTypes { cols: Vec<(String, String, String)> },
360    #[error("duplicated column name references in table {0}: {1:?}")]
361    DuplicatedColumnNames(String, Vec<String>),
362    #[error("{option_name} refers to table not currently being added")]
363    DanglingColumns {
364        option_name: String,
365        items: Vec<UnresolvedItemName>,
366    },
367    #[error("EXCLUDE CONSTRAINTS refers to constraints that do not exist on table {table}")]
368    ConstraintsNotFound {
369        table: String,
370        constraints: Vec<String>,
371    },
372    #[error("Invalid MySQL table reference: {0}")]
373    InvalidTableReference(String),
374    #[error("No tables found for provided reference")]
375    EmptyDatabase,
376    #[error("missing TABLES specification")]
377    RequiresExternalReferences,
378    #[error("No tables found in referenced schemas")]
379    NoTablesFoundForSchemas(Vec<String>),
380    #[error(transparent)]
381    InvalidConnection(#[from] MySqlConnectionValidationError),
382    #[error(
383        "The MySQL system variable 'binlog_row_metadata' is set to an unsupported value: {setting}. Materialize requires this variable to be set to 'FULL' to use the \"CREATE TABLE FROM SOURCE\" syntax for MySQL sources."
384    )]
385    UnsupportedBinlogMetadataSetting { setting: String },
386    #[error(
387        "You are using MySQL version {version}. Materialize requires MySQL 8.0.1 or later to use the \"CREATE TABLE FROM SOURCE\" syntax for MySQL sources."
388    )]
389    UnsupportedMySqlVersion { version: String },
390}
391
392impl MySqlSourcePurificationError {
393    pub fn detail(&self) -> Option<String> {
394        match self {
395            Self::UserLacksPrivileges(missing) => Some(format!(
396                "Missing MySQL privileges: {}",
397                itertools::join(
398                    missing
399                        .iter()
400                        .map(|(privilege, table)| format!("'{}' on '{}'", privilege, table)),
401                    ", "
402                )
403            )),
404            Self::DanglingColumns {
405                option_name: _,
406                items,
407            } => Some(format!(
408                "the following columns are referenced but not added: {}",
409                itertools::join(items, ", ")
410            )),
411            Self::ConstraintsNotFound {
412                table: _,
413                constraints,
414            } => Some(format!(
415                "the following constraints were not found: {}",
416                constraints.join(", ")
417            )),
418            Self::UnrecognizedTypes { cols } => Some(format!(
419                "the following columns contain unsupported types:\n{}",
420                itertools::join(
421                    cols.into_iter().map(|(table, column, data_type)| format!(
422                        "'{}' for {}.{}",
423                        data_type, column, table
424                    )),
425                    "\n"
426                )
427            )),
428            Self::NoTablesFoundForSchemas(schemas) => Some(format!(
429                "missing schemas: {}",
430                itertools::join(schemas.iter(), ", ")
431            )),
432            Self::InvalidConnection(e) => e.detail(),
433            _ => None,
434        }
435    }
436
437    pub fn hint(&self) -> Option<String> {
438        match self {
439            Self::UserSpecifiedDetails => Some(
440                "If trying to use the output of SHOW CREATE SOURCE, remove the DETAILS option."
441                    .into(),
442            ),
443            Self::RequiresExternalReferences => {
444                Some("provide a FOR TABLES (..), FOR SCHEMAS (..), or FOR ALL TABLES clause".into())
445            }
446            Self::InvalidTableReference(_) => Some(
447                "Specify tables names as SCHEMA_NAME.TABLE_NAME in a FOR TABLES (..) clause".into(),
448            ),
449            Self::ConstraintsNotFound { .. } => Some(
450                "Constraint names are matched exactly, including case, against the upstream \
451                 unique index names. The primary key's index is named PRIMARY."
452                    .into(),
453            ),
454            Self::UnrecognizedTypes { cols: _ } => Some(
455                "Check the docs -- some types can be supported using the TEXT COLUMNS option to \
456                ingest their values as text, or ignored using EXCLUDE COLUMNS."
457                    .into(),
458            ),
459            Self::EmptyDatabase => Some(
460                "No tables were found to replicate. This could be because \
461                the user does not have privileges on the intended tables."
462                    .into(),
463            ),
464            Self::UnnecessaryOptionsWithoutReferences(option) => Some(format!(
465                "Remove the {} option, as no tables are being added.",
466                option
467            )),
468            Self::InvalidConnection(e) => e.hint(),
469            _ => None,
470        }
471    }
472}
473
474/// Logical errors detectable during purification for a SQL Server SOURCE.
475#[derive(Debug, Clone, thiserror::Error)]
476pub enum SqlServerSourcePurificationError {
477    #[error("{0} is not a SQL SERVER CONNECTION")]
478    NotSqlServerConnection(FullItemName),
479    #[error("CREATE SOURCE specifies DETAILS option")]
480    UserSpecifiedDetails,
481    #[error("{0} option is unnecessary when no tables are added")]
482    UnnecessaryOptionsWithoutReferences(String),
483    #[error("missing TABLES specification")]
484    RequiresExternalReferences,
485    #[error("{option_name} refers to table not currently being added")]
486    DanglingColumns {
487        option_name: String,
488        items: Vec<UnresolvedItemName>,
489    },
490    #[error("found multiple primary keys for a table. constraints {constraint_names:?}")]
491    MultiplePrimaryKeys { constraint_names: Vec<Arc<str>> },
492    #[error("column {schema_name}.{tbl_name}.{col_name} of type {col_type} is not supported")]
493    UnsupportedColumn {
494        schema_name: Arc<str>,
495        tbl_name: Arc<str>,
496        col_name: Arc<str>,
497        col_type: Arc<str>,
498        context: String,
499    },
500    #[error("Table {tbl_name} had all columns excluded")]
501    AllColumnsExcluded { tbl_name: Arc<str> },
502    #[error("No tables found for provided reference")]
503    NoTables,
504    #[error("programming error: {0}")]
505    ProgrammingError(String),
506    #[error("No start_lsn found for capture instance {0}")]
507    NoStartLsn(String),
508    #[error("Capture instance {capture_instance} has missing columns: {col_names:?}")]
509    CdcMissingColumns {
510        capture_instance: Arc<str>,
511        col_names: Vec<Arc<str>>,
512    },
513}
514
515impl SqlServerSourcePurificationError {
516    pub fn detail(&self) -> Option<String> {
517        match self {
518            Self::DanglingColumns {
519                option_name: _,
520                items,
521            } => Some(format!(
522                "the following columns are referenced but not added: {}",
523                itertools::join(items, ", ")
524            )),
525            Self::UnsupportedColumn { context, .. } => Some(context.clone()),
526            _ => None,
527        }
528    }
529
530    pub fn hint(&self) -> Option<String> {
531        match self {
532            Self::RequiresExternalReferences => {
533                Some("provide a FOR TABLES (..), FOR SCHEMAS (..), or FOR ALL TABLES clause".into())
534            }
535            Self::UnnecessaryOptionsWithoutReferences(option) => Some(format!(
536                "Remove the {} option, as no tables are being added.",
537                option
538            )),
539            Self::NoTables => Some(
540                "No tables were found to replicate. This could be because \
541                the user does not have privileges on the intended tables."
542                    .into(),
543            ),
544            Self::UnsupportedColumn { .. } => {
545                Some("Use EXCLUDE COLUMNS (...) to exclude a column from this source".into())
546            }
547            _ => None,
548        }
549    }
550}
mz_sql/pure/error.rs

mz_sql/pure/
error.rs
