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

mz_sql/pure/
error.rs
