mz_sql/plan/
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::collections::BTreeSet;
11use std::error::Error;
12use std::num::{ParseIntError, TryFromIntError};
13use std::sync::Arc;
14use std::time::Duration;
15use std::{fmt, io};
16
17use itertools::Itertools;
18use mz_expr::EvalError;
19use mz_mysql_util::MySqlError;
20use mz_ore::error::ErrorExt;
21use mz_ore::stack::RecursionLimitError;
22use mz_ore::str::{StrExt, separated};
23use mz_postgres_util::PostgresError;
24use mz_repr::adt::char::InvalidCharLengthError;
25use mz_repr::adt::mz_acl_item::AclMode;
26use mz_repr::adt::numeric::InvalidNumericMaxScaleError;
27use mz_repr::adt::timestamp::InvalidTimestampPrecisionError;
28use mz_repr::adt::varchar::InvalidVarCharMaxLengthError;
29use mz_repr::{CatalogItemId, ColumnName, strconv};
30use mz_sql_parser::ast::display::AstDisplay;
31use mz_sql_parser::ast::{IdentError, UnresolvedItemName};
32use mz_sql_parser::parser::{ParserError, ParserStatementError};
33use mz_sql_server_util::SqlServerError;
34use mz_storage_types::connections::InvalidAwsPrivatelinkServiceName;
35use mz_storage_types::sources::ExternalReferenceResolutionError;
36
37use crate::catalog::{
38    CatalogError, CatalogItemType, ErrorMessageObjectDescription, SystemObjectType,
39};
40use crate::names::{PartialItemName, ResolvedItemName};
41use crate::plan::ObjectType;
42use crate::plan::plan_utils::JoinSide;
43use crate::plan::scope::ScopeItem;
44use crate::plan::typeconv::CastContext;
45use crate::pure::error::{
46    CsrPurificationError, GluePurificationError, IcebergSinkPurificationError,
47    KafkaSinkPurificationError, KafkaSourcePurificationError, LoadGeneratorSourcePurificationError,
48    MySqlSourcePurificationError, PgSourcePurificationError, SqlServerSourcePurificationError,
49};
50use crate::session::vars::VarError;
51
52#[derive(Debug)]
53pub enum PlanError {
54    /// This feature is not yet supported, but may be supported at some point in the future.
55    Unsupported {
56        feature: String,
57        discussion_no: Option<usize>,
58    },
59    /// A burst `HYDRATION SIZE` equal to the cluster `SIZE` (a no-op burst).
60    HydrationSizeEqualsClusterSize {
61        size: String,
62    },
63    /// This feature is not supported, and will likely never be supported.
64    NeverSupported {
65        feature: String,
66        documentation_link: Option<String>,
67        details: Option<String>,
68    },
69    UnknownColumn {
70        table: Option<PartialItemName>,
71        column: ColumnName,
72        similar: Box<[ColumnName]>,
73    },
74    UngroupedColumn {
75        table: Option<PartialItemName>,
76        column: ColumnName,
77    },
78    ItemWithoutColumns {
79        name: String,
80        item_type: CatalogItemType,
81    },
82    WrongJoinTypeForLateralColumn {
83        table: Option<PartialItemName>,
84        column: ColumnName,
85    },
86    AmbiguousColumn(ColumnName),
87    TooManyColumns {
88        max_num_columns: usize,
89        req_num_columns: usize,
90    },
91    ColumnAlreadyExists {
92        column_name: ColumnName,
93        object_name: String,
94    },
95    AmbiguousTable(PartialItemName),
96    UnknownColumnInUsingClause {
97        column: ColumnName,
98        join_side: JoinSide,
99    },
100    AmbiguousColumnInUsingClause {
101        column: ColumnName,
102        join_side: JoinSide,
103    },
104    MisqualifiedName(String),
105    OverqualifiedDatabaseName(String),
106    OverqualifiedSchemaName(String),
107    UnderqualifiedColumnName(String),
108    SubqueriesDisallowed {
109        context: String,
110    },
111    UnknownParameter(usize),
112    ParameterNotAllowed(String),
113    WrongParameterType(usize, String, String),
114    RecursionLimit(RecursionLimitError),
115    StrconvParse(strconv::ParseError),
116    Catalog(CatalogError),
117    UpsertSinkWithoutKey,
118    UpsertSinkWithInvalidKey {
119        name: String,
120        desired_key: Vec<String>,
121        valid_keys: Vec<Vec<String>>,
122    },
123    IcebergSinkUnsupportedKeyType {
124        column: String,
125        column_type: String,
126    },
127    InvalidWmrRecursionLimit(String),
128    InvalidNumericMaxScale(InvalidNumericMaxScaleError),
129    InvalidCharLength(InvalidCharLengthError),
130    InvalidId(CatalogItemId),
131    InvalidIdent(IdentError),
132    InvalidObject(Box<ResolvedItemName>),
133    InvalidObjectType {
134        expected_type: SystemObjectType,
135        actual_type: SystemObjectType,
136        object_name: String,
137    },
138    InvalidPrivilegeTypes {
139        invalid_privileges: AclMode,
140        object_description: ErrorMessageObjectDescription,
141    },
142    InvalidVarCharMaxLength(InvalidVarCharMaxLengthError),
143    InvalidTimestampPrecision(InvalidTimestampPrecisionError),
144    InvalidSecret(Box<ResolvedItemName>),
145    InvalidTemporarySchema,
146    InvalidCast {
147        name: String,
148        ccx: CastContext,
149        from: String,
150        to: String,
151    },
152    /// Range type with an element type that is not supported (e.g. float, uint).
153    UnsupportedRangeElementType {
154        element_type_name: String,
155    },
156    InvalidTable {
157        name: String,
158    },
159    InvalidVersion {
160        name: String,
161        version: String,
162    },
163    InvalidSinkFrom {
164        name: String,
165        item_type: String,
166    },
167    InvalidDependency {
168        name: String,
169        item_type: String,
170    },
171    MangedReplicaName(String),
172    ParserStatement(ParserStatementError),
173    Parser(ParserError),
174    DropViewOnMaterializedView(String),
175    DependentObjectsStillExist {
176        object_type: String,
177        object_name: String,
178        // (dependent type, name)
179        dependents: Vec<(String, String)>,
180    },
181    AlterViewOnMaterializedView(String),
182    ShowCreateViewOnMaterializedView(String),
183    ExplainViewOnMaterializedView(String),
184    UnacceptableTimelineName(String),
185    FetchingCsrSchemaFailed {
186        schema_lookup: String,
187        cause: Arc<dyn Error + Send + Sync>,
188    },
189    PostgresConnectionErr {
190        cause: Arc<mz_postgres_util::PostgresError>,
191    },
192    MySqlConnectionErr {
193        cause: Arc<MySqlError>,
194    },
195    SqlServerConnectionErr {
196        cause: Arc<SqlServerError>,
197    },
198    SubsourceNameConflict {
199        name: UnresolvedItemName,
200        upstream_references: Vec<UnresolvedItemName>,
201    },
202    SubsourceDuplicateReference {
203        name: UnresolvedItemName,
204        target_names: Vec<UnresolvedItemName>,
205    },
206    NoTablesFoundForSchemas(Vec<String>),
207    InvalidProtobufSchema {
208        cause: protobuf_native::OperationFailedError,
209    },
210    InvalidOptionValue {
211        // Expected to be generated from the `to_ast_string` value on the option
212        // name.
213        option_name: String,
214        err: Box<PlanError>,
215    },
216    UnexpectedDuplicateReference {
217        name: UnresolvedItemName,
218    },
219    /// Declaration of a recursive type did not match the inferred type.
220    RecursiveTypeMismatch(String, Vec<String>, Vec<String>),
221    UnknownFunction {
222        name: String,
223        arg_types: Vec<String>,
224    },
225    IndistinctFunction {
226        name: String,
227        arg_types: Vec<String>,
228    },
229    UnknownOperator {
230        name: String,
231        arg_types: Vec<String>,
232    },
233    IndistinctOperator {
234        name: String,
235        arg_types: Vec<String>,
236    },
237    InvalidPrivatelinkAvailabilityZone {
238        name: String,
239        supported_azs: BTreeSet<String>,
240    },
241    InvalidPrivatelinkServiceName(InvalidAwsPrivatelinkServiceName),
242    DuplicatePrivatelinkAvailabilityZone {
243        duplicate_azs: BTreeSet<String>,
244    },
245    InvalidSchemaName,
246    ItemAlreadyExists {
247        name: String,
248        item_type: CatalogItemType,
249    },
250    ManagedCluster {
251        cluster_name: String,
252    },
253    InvalidKeysInSubscribeEnvelopeUpsert,
254    InvalidKeysInSubscribeEnvelopeDebezium,
255    DuplicateKeyColumnInSubscribeEnvelope {
256        column_name: String,
257    },
258    InvalidPartitionByEnvelopeDebezium {
259        column_name: String,
260    },
261    InvalidOrderByInSubscribeWithinTimestampOrderBy,
262    FromValueRequiresParen,
263    VarError(VarError),
264    UnsolvablePolymorphicFunctionInput,
265    ShowCommandInView,
266    WebhookValidationDoesNotUseColumns,
267    WebhookValidationNonDeterministic,
268    InternalFunctionCall,
269    CommentTooLong {
270        length: usize,
271        max_size: usize,
272    },
273    InvalidTimestampInterval {
274        min: Duration,
275        max: Duration,
276        requested: Duration,
277    },
278    InvalidGroupSizeHints,
279    PgSourcePurification(PgSourcePurificationError),
280    KafkaSourcePurification(KafkaSourcePurificationError),
281    KafkaSinkPurification(KafkaSinkPurificationError),
282    IcebergSinkPurification(IcebergSinkPurificationError),
283    LoadGeneratorSourcePurification(LoadGeneratorSourcePurificationError),
284    CsrPurification(CsrPurificationError),
285    GluePurification(GluePurificationError),
286    MySqlSourcePurification(MySqlSourcePurificationError),
287    SqlServerSourcePurificationError(SqlServerSourcePurificationError),
288    UseTablesForSources(String),
289    MissingName(CatalogItemType),
290    InvalidRefreshAt,
291    InvalidRefreshEveryAlignedTo,
292    MismatchedObjectType {
293        name: PartialItemName,
294        is_type: ObjectType,
295        expected_type: ObjectType,
296    },
297    /// MZ failed to generate cast for the data type.
298    TableContainsUningestableTypes {
299        name: String,
300        type_: String,
301        column: String,
302    },
303    RetainHistoryLow {
304        limit: Duration,
305    },
306    RetainHistoryRequired,
307    UntilReadyTimeoutRequired,
308    SubsourceResolutionError(ExternalReferenceResolutionError),
309    Replan(String),
310    Internal(String),
311    NetworkPolicyLockoutError,
312    NetworkPolicyInUse,
313    /// Expected a constant expression that evaluates without an error to a non-null value.
314    ConstantExpressionSimplificationFailed(String),
315    InvalidLimit(String),
316    InvalidOffset(String),
317    /// The named cursor does not exist.
318    UnknownCursor(String),
319    CopyFromTargetTableDropped {
320        target_name: String,
321    },
322    /// AS OF or UP TO should be an expression that is castable and simplifiable to a non-null mz_timestamp value.
323    InvalidAsOfUpTo,
324    InvalidReplacement {
325        item_type: CatalogItemType,
326        item_name: PartialItemName,
327        replacement_type: CatalogItemType,
328        replacement_name: PartialItemName,
329    },
330    // TODO(benesch): eventually all errors should be structured.
331    Unstructured(String),
332}
333
334impl PlanError {
335    pub(crate) fn ungrouped_column(item: &ScopeItem) -> PlanError {
336        PlanError::UngroupedColumn {
337            table: item.table_name.clone(),
338            column: item.column_name.clone(),
339        }
340    }
341
342    pub fn detail(&self) -> Option<String> {
343        match self {
344            Self::HydrationSizeEqualsClusterSize { .. } => Some(
345                "A burst replica at the same size as the steady replicas would not \
346                 accelerate hydration."
347                    .into(),
348            ),
349            Self::NeverSupported { details, .. } => details.clone(),
350            Self::FetchingCsrSchemaFailed { cause, .. } => Some(cause.to_string_with_causes()),
351            Self::PostgresConnectionErr { cause } => Some(cause.to_string_with_causes()),
352            Self::InvalidProtobufSchema { cause } => Some(cause.to_string_with_causes()),
353            Self::InvalidOptionValue { err, .. } => err.detail(),
354            Self::UpsertSinkWithInvalidKey {
355                name,
356                desired_key,
357                valid_keys,
358            } => {
359                let valid_keys = if valid_keys.is_empty() {
360                    "There are no known valid unique keys for the underlying relation.".into()
361                } else {
362                    format!(
363                        "The following keys are known to be unique for the underlying relation:\n{}",
364                        valid_keys
365                            .iter()
366                            .map(|k|
367                                format!("  ({})", k.iter().map(|c| c.as_str().quoted()).join(", "))
368                            )
369                            .join("\n"),
370                    )
371                };
372                Some(format!(
373                    "Materialize could not prove that the specified upsert envelope key ({}) \
374                    was a unique key of the underlying relation {}. {valid_keys}",
375                    separated(", ", desired_key.iter().map(|c| c.as_str().quoted())),
376                    name.quoted()
377                ))
378            }
379            Self::VarError(e) => e.detail(),
380            Self::InternalFunctionCall => Some("This function is for the internal use of the database system and cannot be called directly.".into()),
381            Self::PgSourcePurification(e) => e.detail(),
382            Self::MySqlSourcePurification(e) => e.detail(),
383            Self::SqlServerSourcePurificationError(e) => e.detail(),
384            Self::KafkaSourcePurification(e) => e.detail(),
385            Self::LoadGeneratorSourcePurification(e) => e.detail(),
386            Self::CsrPurification(e) => e.detail(),
387            Self::GluePurification(e) => e.detail(),
388            Self::KafkaSinkPurification(e) => e.detail(),
389            Self::IcebergSinkPurification(e) => e.detail(),
390            Self::SubsourceNameConflict {
391                name: _,
392                upstream_references,
393            } => Some(format!(
394                "referenced tables with duplicate name: {}",
395                itertools::join(upstream_references, ", ")
396            )),
397            Self::SubsourceDuplicateReference {
398                name: _,
399                target_names,
400            } => Some(format!(
401                "subsources referencing table: {}",
402                itertools::join(target_names, ", ")
403            )),
404            Self::InvalidPartitionByEnvelopeDebezium { .. } => Some(
405                "When using ENVELOPE DEBEZIUM, only columns in the key can be referenced in the PARTITION BY expression.".to_string()
406            ),
407            Self::NoTablesFoundForSchemas(schemas) => Some(format!(
408                "missing schemas: {}",
409                separated(", ", schemas.iter().map(|c| c.quoted()))
410            )),
411            _ => None,
412        }
413    }
414
415    pub fn hint(&self) -> Option<String> {
416        match self {
417            Self::DropViewOnMaterializedView(_) => {
418                Some("Use DROP MATERIALIZED VIEW to remove a materialized view.".into())
419            }
420            Self::DependentObjectsStillExist {..} => Some("Use DROP ... CASCADE to drop the dependent objects too.".into()),
421            Self::AlterViewOnMaterializedView(_) => {
422                Some("Use ALTER MATERIALIZED VIEW to rename a materialized view.".into())
423            }
424            Self::ShowCreateViewOnMaterializedView(_) => {
425                Some("Use SHOW CREATE MATERIALIZED VIEW to show a materialized view.".into())
426            }
427            Self::ExplainViewOnMaterializedView(_) => {
428                Some("Use EXPLAIN [...] MATERIALIZED VIEW to explain a materialized view.".into())
429            }
430            Self::UnacceptableTimelineName(_) => {
431                Some("The prefix \"mz_\" is reserved for system timelines.".into())
432            }
433            Self::PostgresConnectionErr { cause } => {
434                if let Some(cause) = cause.source() {
435                    if let Some(cause) = cause.downcast_ref::<io::Error>() {
436                        if cause.kind() == io::ErrorKind::TimedOut {
437                            return Some(
438                                "Do you have a firewall or security group that is \
439                                preventing Materialize from connecting to your PostgreSQL server?"
440                                    .into(),
441                            );
442                        }
443                    }
444                }
445                None
446            }
447            Self::InvalidOptionValue { err, .. } => err.hint(),
448            Self::UnknownFunction { ..} => Some("No function matches the given name and argument types.  You might need to add explicit type casts.".into()),
449            Self::IndistinctFunction {..} => {
450                Some("Could not choose a best candidate function.  You might need to add explicit type casts.".into())
451            }
452            Self::UnknownOperator {..} => {
453                Some("No operator matches the given name and argument types.  You might need to add explicit type casts.".into())
454            }
455            Self::IndistinctOperator {..} => {
456                Some("Could not choose a best candidate operator.  You might need to add explicit type casts.".into())
457            },
458            Self::InvalidPrivatelinkAvailabilityZone { supported_azs, ..} => {
459                let supported_azs_str = supported_azs.iter().join("\n  ");
460                Some(format!("Did you supply an availability zone name instead of an ID? Known availability zone IDs:\n  {}", supported_azs_str))
461            }
462            Self::InvalidPrivatelinkServiceName(err) => Some(err.hint()),
463            Self::DuplicatePrivatelinkAvailabilityZone { duplicate_azs, ..} => {
464                let duplicate_azs  = duplicate_azs.iter().join("\n  ");
465                Some(format!("Duplicated availability zones:\n  {}", duplicate_azs))
466            }
467            Self::InvalidKeysInSubscribeEnvelopeUpsert => {
468                Some("All keys must be columns on the underlying relation.".into())
469            }
470            Self::InvalidKeysInSubscribeEnvelopeDebezium => {
471                Some("All keys must be columns on the underlying relation.".into())
472            }
473            Self::DuplicateKeyColumnInSubscribeEnvelope { .. } => {
474                Some("Each KEY column must be listed at most once.".into())
475            }
476            Self::InvalidOrderByInSubscribeWithinTimestampOrderBy => {
477                Some("All order bys must be output columns.".into())
478            }
479            Self::UpsertSinkWithInvalidKey { .. } | Self::UpsertSinkWithoutKey => {
480                Some("See: https://materialize.com/s/sink-key-selection".into())
481            }
482            Self::IcebergSinkUnsupportedKeyType { .. } => {
483                Some("Iceberg equality delete keys must be primitive, non-floating-point columns.".into())
484            }
485            Self::Catalog(e) => e.hint(),
486            Self::VarError(e) => e.hint(),
487            Self::PgSourcePurification(e) => e.hint(),
488            Self::MySqlSourcePurification(e) => e.hint(),
489            Self::SqlServerSourcePurificationError(e) => e.hint(),
490            Self::KafkaSourcePurification(e) => e.hint(),
491            Self::LoadGeneratorSourcePurification(e) => e.hint(),
492            Self::CsrPurification(e) => e.hint(),
493            Self::GluePurification(e) => e.hint(),
494            Self::KafkaSinkPurification(e) => e.hint(),
495            Self::UnknownColumn { table, similar, .. } => {
496                let suffix = "Make sure to surround case sensitive names in double quotes.";
497                match &similar[..] {
498                    [] => None,
499                    [column] => Some(format!("The similarly named column {} does exist. {suffix}", ColumnDisplay { table, column })),
500                    names => {
501                        let similar = names.into_iter().map(|column| ColumnDisplay { table, column }).join(", ");
502                        Some(format!("There are similarly named columns that do exist: {similar}. {suffix}"))
503                    }
504                }
505            }
506            Self::RecursiveTypeMismatch(..) => {
507                Some("You will need to rewrite or cast the query's expressions.".into())
508            },
509            Self::InvalidRefreshAt
510            | Self::InvalidRefreshEveryAlignedTo => {
511                Some("Calling `mz_now()` is allowed.".into())
512            },
513            Self::TableContainsUningestableTypes { column,.. } => {
514                Some(format!("Remove the table or use TEXT COLUMNS ({column}, ..) to ingest this column as text"))
515            }
516            Self::RetainHistoryLow { .. } | Self::RetainHistoryRequired => {
517                Some("Use ALTER ... RESET (RETAIN HISTORY) to set the retain history to its default and lowest value.".into())
518            }
519            Self::NetworkPolicyInUse => {
520                Some("Use ALTER SYSTEM SET 'network_policy' to change the default network policy.".into())
521            }
522            Self::WrongParameterType(_, _, _) => {
523                Some("EXECUTE automatically inserts only such casts that are allowed in an assignment cast context.  Try adding an explicit cast.".into())
524            }
525            Self::InvalidSchemaName => {
526                Some("Use SET schema = name to select a schema.  Use SHOW SCHEMAS to list available schemas.  Use SHOW search_path to show the schema names that we looked for, but none of them existed.".into())
527            }
528            _ => None,
529        }
530    }
531}
532
533impl fmt::Display for PlanError {
534    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
535        match self {
536            Self::Unsupported { feature, discussion_no } => {
537                write!(f, "{} not yet supported", feature)?;
538                if let Some(discussion_no) = discussion_no {
539                    write!(f, ", see https://github.com/MaterializeInc/materialize/discussions/{} for more details", discussion_no)?;
540                }
541                Ok(())
542            }
543            Self::HydrationSizeEqualsClusterSize { size } => {
544                write!(f, "HYDRATION SIZE must differ from the cluster SIZE ('{size}')")
545            }
546            Self::NeverSupported { feature, documentation_link: documentation_path,.. } => {
547                write!(f, "{feature} is not supported",)?;
548                if let Some(documentation_path) = documentation_path {
549                    write!(f, ", for more information consult the documentation at https://materialize.com/docs/{documentation_path}")?;
550                }
551                Ok(())
552            }
553            Self::UnknownColumn { table, column, similar: _ } => write!(
554                f,
555                "column {} does not exist",
556                ColumnDisplay { table, column }
557            ),
558            Self::UngroupedColumn { table, column } => write!(
559                f,
560                "column {} must appear in the GROUP BY clause or be used in an aggregate function",
561                ColumnDisplay { table, column },
562            ),
563            Self::ItemWithoutColumns { name, item_type } => {
564                let name = name.quoted();
565                write!(f, "{item_type} {name} does not have columns")
566            }
567            Self::WrongJoinTypeForLateralColumn { table, column } => write!(
568                f,
569                "column {} cannot be referenced from this part of the query: \
570                the combining JOIN type must be INNER or LEFT for a LATERAL reference",
571                ColumnDisplay { table, column },
572            ),
573            Self::AmbiguousColumn(column) => write!(
574                f,
575                "column reference {} is ambiguous",
576                column.quoted()
577            ),
578            Self::TooManyColumns { max_num_columns, req_num_columns } => write!(
579                f,
580                "attempt to create relation with too many columns, {} max: {}",
581                req_num_columns, max_num_columns
582            ),
583            Self::ColumnAlreadyExists { column_name, object_name } => write!(
584                f,
585                "column {} of relation {} already exists",
586                column_name.quoted(), object_name.quoted(),
587            ),
588            Self::AmbiguousTable(table) => write!(
589                f,
590                "table reference {} is ambiguous",
591                table.item.as_str().quoted()
592            ),
593            Self::UnknownColumnInUsingClause { column, join_side } => write!(
594                f,
595                "column {} specified in USING clause does not exist in {} table",
596                column.quoted(),
597                join_side,
598            ),
599            Self::AmbiguousColumnInUsingClause { column, join_side } => write!(
600                f,
601                "common column name {} appears more than once in {} table",
602                column.quoted(),
603                join_side,
604            ),
605            Self::MisqualifiedName(name) => write!(
606                f,
607                "qualified name did not have between 1 and 3 components: {}",
608                name
609            ),
610            Self::OverqualifiedDatabaseName(name) => write!(
611                f,
612                "database name '{}' does not have exactly one component",
613                name
614            ),
615            Self::OverqualifiedSchemaName(name) => write!(
616                f,
617                "schema name '{}' cannot have more than two components",
618                name
619            ),
620            Self::UnderqualifiedColumnName(name) => write!(
621                f,
622                "column name '{}' must have at least a table qualification",
623                name
624            ),
625            Self::UnacceptableTimelineName(name) => {
626                write!(f, "unacceptable timeline name {}", name.quoted(),)
627            }
628            Self::SubqueriesDisallowed { context } => {
629                write!(f, "{} does not allow subqueries", context)
630            }
631            Self::UnknownParameter(n) => write!(f, "there is no parameter ${}", n),
632            Self::ParameterNotAllowed(object_type) => write!(f, "{} cannot have parameters", object_type),
633            Self::WrongParameterType(i, expected_ty, actual_ty) => write!(f, "unable to cast given parameter ${}: expected {}, got {}", i, expected_ty, actual_ty),
634            Self::RecursionLimit(e) => write!(f, "{}", e),
635            Self::StrconvParse(e) => write!(f, "{}", e),
636            Self::Catalog(e) => write!(f, "{}", e),
637            Self::UpsertSinkWithoutKey => write!(f, "upsert sinks must specify a key"),
638            Self::UpsertSinkWithInvalidKey { .. } => {
639                write!(f, "upsert key could not be validated as unique")
640            }
641            Self::IcebergSinkUnsupportedKeyType { column, column_type } => {
642                write!(f, "column {column} has type {column_type} which cannot be used as an Iceberg equality delete key")
643            }
644            Self::InvalidWmrRecursionLimit(msg) => write!(f, "Invalid WITH MUTUALLY RECURSIVE recursion limit. {}", msg),
645            Self::InvalidNumericMaxScale(e) => e.fmt(f),
646            Self::InvalidCharLength(e) => e.fmt(f),
647            Self::InvalidVarCharMaxLength(e) => e.fmt(f),
648            Self::InvalidTimestampPrecision(e) => e.fmt(f),
649            Self::Parser(e) => e.fmt(f),
650            Self::ParserStatement(e) => e.fmt(f),
651            Self::Unstructured(e) => write!(f, "{}", e),
652            Self::InvalidId(id) => write!(f, "invalid id {}", id),
653            Self::InvalidIdent(err) => write!(f, "invalid identifier, {err}"),
654            Self::InvalidObject(i) => write!(f, "{} is not a database object", i.full_name_str()),
655            Self::InvalidObjectType{expected_type, actual_type, object_name} => write!(f, "{actual_type} {object_name} is not a {expected_type}"),
656            Self::InvalidPrivilegeTypes{ invalid_privileges, object_description, } => {
657                write!(f, "invalid privilege types {} for {}", invalid_privileges.to_error_string(), object_description)
658            },
659            Self::InvalidSecret(i) => write!(f, "{} is not a secret", i.full_name_str()),
660            Self::InvalidTemporarySchema => {
661                write!(f, "cannot create temporary item in non-temporary schema")
662            }
663            Self::InvalidCast { name, ccx, from, to } =>{
664                write!(
665                    f,
666                    "{name} does not support {ccx}casting from {from} to {to}",
667                    ccx = if matches!(ccx, CastContext::Implicit) {
668                        "implicitly "
669                    } else {
670                        ""
671                    },
672                )
673            }
674            Self::UnsupportedRangeElementType { element_type_name } => {
675                write!(f, "range type over {} is not supported", element_type_name)
676            }
677            Self::InvalidTable { name } => {
678                write!(f, "invalid table definition for {}", name.quoted())
679            },
680            Self::InvalidVersion { name, version } => {
681                write!(f, "invalid version {} for {}", version.quoted(), name.quoted())
682            },
683            Self::InvalidSinkFrom { name, item_type } => {
684                write!(f, "{item_type} {name} cannot be exported as a sink")
685            },
686            Self::InvalidDependency { name, item_type } => {
687                write!(f, "{item_type} {name} cannot be depended upon")
688            },
689            Self::DropViewOnMaterializedView(name)
690            | Self::AlterViewOnMaterializedView(name)
691            | Self::ShowCreateViewOnMaterializedView(name)
692            | Self::ExplainViewOnMaterializedView(name) => write!(f, "{name} is not a view"),
693            Self::FetchingCsrSchemaFailed { schema_lookup, .. } => {
694                write!(f, "failed to fetch schema {schema_lookup} from schema registry")
695            }
696            Self::PostgresConnectionErr { .. } => {
697                write!(f, "failed to connect to PostgreSQL database")
698            }
699            Self::MySqlConnectionErr { cause } => {
700                write!(f, "failed to connect to MySQL database: {}", cause)
701            }
702            Self::SqlServerConnectionErr { cause } => {
703                write!(f, "failed to connect to SQL Server database: {}", cause)
704            }
705            Self::SubsourceNameConflict {
706                name , upstream_references: _,
707            } => {
708                write!(f, "multiple subsources would be named {}", name)
709            },
710            Self::SubsourceDuplicateReference {
711                name,
712                target_names: _,
713            } => {
714                write!(f, "multiple subsources refer to table {}", name)
715            },
716            Self::NoTablesFoundForSchemas(schemas) => {
717                write!(f, "no tables found in referenced schemas: {}",
718                    separated(", ", schemas.iter().map(|c| c.quoted()))
719                )
720            },
721            Self::InvalidProtobufSchema { .. } => {
722                write!(f, "invalid protobuf schema")
723            }
724            Self::DependentObjectsStillExist {object_type, object_name, dependents} => {
725                let reason = match &dependents[..] {
726                    [] => " because other objects depend on it".to_string(),
727                    dependents => {
728                        let dependents = dependents.iter().map(|(dependent_type, dependent_name)| format!("{} {}", dependent_type, dependent_name.quoted())).join(", ");
729                        format!(": still depended upon by {dependents}")
730                    },
731                };
732                let object_name = object_name.quoted();
733                write!(f, "cannot drop {object_type} {object_name}{reason}")
734            }
735            Self::InvalidOptionValue { option_name, err } => write!(f, "invalid {} option value: {}", option_name, err),
736            Self::UnexpectedDuplicateReference { name } => write!(f, "unexpected multiple references to {}", name.to_ast_string_simple()),
737            Self::RecursiveTypeMismatch(name, declared, inferred) => {
738                let declared = separated(", ", declared);
739                let inferred = separated(", ", inferred);
740                let name = name.quoted();
741                write!(f, "WITH MUTUALLY RECURSIVE query {name} declared types ({declared}), but query returns types ({inferred})")
742            },
743            Self::UnknownFunction {name, arg_types, ..} => {
744                write!(f, "function {}({}) does not exist", name, arg_types.join(", "))
745            },
746            Self::IndistinctFunction {name, arg_types, ..} => {
747                write!(f, "function {}({}) is not unique", name, arg_types.join(", "))
748            },
749            Self::UnknownOperator {name, arg_types, ..} => {
750                write!(f, "operator does not exist: {}", match arg_types.as_slice(){
751                    [typ] => format!("{} {}", name, typ),
752                    [ltyp, rtyp] => {
753                        format!("{} {} {}", ltyp, name, rtyp)
754                    }
755                    _ => unreachable!("non-unary non-binary operator"),
756                })
757            },
758            Self::IndistinctOperator {name, arg_types, ..} => {
759                write!(f, "operator is not unique: {}", match arg_types.as_slice(){
760                    [typ] => format!("{} {}", name, typ),
761                    [ltyp, rtyp] => {
762                        format!("{} {} {}", ltyp, name, rtyp)
763                    }
764                    _ => unreachable!("non-unary non-binary operator"),
765                })
766            },
767            Self::InvalidPrivatelinkAvailabilityZone { name, ..} => write!(f, "invalid AWS PrivateLink availability zone {}", name.quoted()),
768            Self::InvalidPrivatelinkServiceName(err) => err.fmt(f),
769            Self::DuplicatePrivatelinkAvailabilityZone {..} =>   write!(f, "connection cannot contain duplicate availability zones"),
770            Self::InvalidSchemaName => write!(f, "no valid schema selected"),
771            Self::ItemAlreadyExists { name, item_type } => write!(f, "{item_type} {} already exists", name.quoted()),
772            Self::ManagedCluster {cluster_name} => write!(f, "cannot modify managed cluster {cluster_name}"),
773            Self::InvalidKeysInSubscribeEnvelopeUpsert => {
774                write!(f, "invalid keys in SUBSCRIBE ENVELOPE UPSERT (KEY (..))")
775            }
776            Self::InvalidKeysInSubscribeEnvelopeDebezium => {
777                write!(f, "invalid keys in SUBSCRIBE ENVELOPE DEBEZIUM (KEY (..))")
778            }
779            Self::DuplicateKeyColumnInSubscribeEnvelope { column_name } => {
780                write!(
781                    f,
782                    "column {} appears more than once in SUBSCRIBE ENVELOPE KEY clause",
783                    column_name.quoted(),
784                )
785            }
786            Self::InvalidPartitionByEnvelopeDebezium { column_name } => {
787                write!(
788                    f,
789                    "PARTITION BY expression cannot refer to non-key column {}",
790                    column_name.quoted(),
791                )
792            }
793            Self::InvalidOrderByInSubscribeWithinTimestampOrderBy => {
794                write!(f, "invalid ORDER BY in SUBSCRIBE WITHIN TIMESTAMP ORDER BY")
795            }
796            Self::FromValueRequiresParen => f.write_str(
797                "VALUES expression in FROM clause must be surrounded by parentheses"
798            ),
799            Self::VarError(e) => e.fmt(f),
800            Self::UnsolvablePolymorphicFunctionInput => f.write_str(
801                "could not determine polymorphic type because input has type unknown"
802            ),
803            Self::ShowCommandInView => f.write_str("SHOW commands are not allowed in views"),
804            Self::WebhookValidationDoesNotUseColumns => f.write_str(
805                "expression provided in CHECK does not reference any columns"
806            ),
807            Self::WebhookValidationNonDeterministic => f.write_str(
808                "expression provided in CHECK is not deterministic"
809            ),
810            Self::InternalFunctionCall => f.write_str("cannot call function with arguments of type internal"),
811            Self::CommentTooLong { length, max_size } => {
812                write!(f, "provided comment was {length} bytes long, max size is {max_size} bytes")
813            }
814            Self::InvalidTimestampInterval { min, max, requested } => {
815                write!(f, "invalid timestamp interval of {}ms, must be in the range [{}ms, {}ms]", requested.as_millis(), min.as_millis(), max.as_millis())
816            }
817            Self::InvalidGroupSizeHints => f.write_str("EXPECTED GROUP SIZE cannot be provided \
818                simultaneously with any of AGGREGATE INPUT GROUP SIZE, DISTINCT ON INPUT GROUP SIZE, \
819                or LIMIT INPUT GROUP SIZE"),
820            Self::PgSourcePurification(e) => write!(f, "POSTGRES source validation: {}", e),
821            Self::KafkaSourcePurification(e) => write!(f, "KAFKA source validation: {}", e),
822            Self::LoadGeneratorSourcePurification(e) => write!(f, "LOAD GENERATOR source validation: {}", e),
823            Self::KafkaSinkPurification(e) => write!(f, "KAFKA sink validation: {}", e),
824            Self::IcebergSinkPurification(e) => write!(f, "ICEBERG sink validation: {}", e),
825            Self::CsrPurification(e) => write!(f, "CONFLUENT SCHEMA REGISTRY validation: {}", e),
826            Self::GluePurification(e) => write!(f, "AWS GLUE SCHEMA REGISTRY validation: {}", e),
827            Self::MySqlSourcePurification(e) => write!(f, "MYSQL source validation: {}", e),
828            Self::SqlServerSourcePurificationError(e) => write!(f, "SQL SERVER source validation: {}", e),
829            Self::UseTablesForSources(command) => write!(f, "{command} not supported; use CREATE TABLE .. FROM SOURCE instead"),
830            Self::MangedReplicaName(name) => {
831                write!(f, "{name} is reserved for replicas of managed clusters")
832            }
833            Self::MissingName(item_type) => {
834                write!(f, "unspecified name for {item_type}")
835            }
836            Self::InvalidRefreshAt => {
837                write!(f, "REFRESH AT argument must be an expression that can be simplified \
838                           and/or cast to a constant whose type is mz_timestamp")
839            }
840            Self::InvalidRefreshEveryAlignedTo => {
841                write!(f, "REFRESH EVERY ... ALIGNED TO argument must be an expression that can be simplified \
842                           and/or cast to a constant whose type is mz_timestamp")
843            }
844            Self::MismatchedObjectType {
845                name,
846                is_type,
847                expected_type,
848            } => {
849                write!(
850                    f,
851                    "{name} is {} {} not {} {}",
852                    if *is_type == ObjectType::Index {
853                        "an"
854                    } else {
855                        "a"
856                    },
857                    is_type.to_string().to_lowercase(),
858                    if *expected_type == ObjectType::Index {
859                        "an"
860                    } else {
861                        "a"
862                    },
863                    expected_type.to_string().to_lowercase()
864                )
865            }
866            Self::TableContainsUningestableTypes { name, type_, column } => {
867                write!(f, "table {name} contains column {column} of type {type_} which Materialize cannot currently ingest")
868            },
869            Self::RetainHistoryLow { limit } => {
870                write!(f, "RETAIN HISTORY cannot be set lower than {}ms", limit.as_millis())
871            },
872            Self::RetainHistoryRequired => {
873                write!(f, "RETAIN HISTORY cannot be disabled or set to 0")
874            },
875            Self::SubsourceResolutionError(e) => write!(f, "{}", e),
876            Self::Replan(msg) => write!(f, "internal error while replanning, please contact support: {msg}"),
877            Self::Internal(msg) => write!(f, "internal error: {msg}"),
878            Self::NetworkPolicyLockoutError => write!(f, "policy would block current session IP"),
879            Self::NetworkPolicyInUse => write!(f, "network policy is currently in use"),
880            Self::UntilReadyTimeoutRequired => {
881                write!(f, "TIMEOUT=<duration> option is required for ALTER CLUSTER ... WITH (WAIT UNTIL READY ( ... ))")
882            },
883            Self::ConstantExpressionSimplificationFailed(e) => write!(f, "{}", e),
884            Self::InvalidLimit(e) => write!(f, "Invalid LIMIT clause: {}", e),
885            Self::InvalidOffset(e) => write!(f, "Invalid OFFSET clause: {}", e),
886            Self::UnknownCursor(name) => {
887                write!(f, "cursor {} does not exist", name.quoted())
888            }
889            Self::CopyFromTargetTableDropped { target_name: name } => {
890                write!(f, "COPY FROM's target table {} was dropped", name.quoted())
891            }
892            Self::InvalidAsOfUpTo => write!(
893                f,
894                "AS OF or UP TO should be castable to a (non-null) mz_timestamp value",
895            ),
896            Self::InvalidReplacement {
897                item_type, item_name, replacement_type, replacement_name,
898            } => {
899                write!(
900                    f,
901                    "cannot replace {item_type} {item_name} \
902                     with {replacement_type} {replacement_name}",
903                )
904            }
905        }
906    }
907}
908
909impl Error for PlanError {}
910
911impl From<CatalogError> for PlanError {
912    fn from(e: CatalogError) -> PlanError {
913        PlanError::Catalog(e)
914    }
915}
916
917impl From<strconv::ParseError> for PlanError {
918    fn from(e: strconv::ParseError) -> PlanError {
919        PlanError::StrconvParse(e)
920    }
921}
922
923impl From<RecursionLimitError> for PlanError {
924    fn from(e: RecursionLimitError) -> PlanError {
925        PlanError::RecursionLimit(e)
926    }
927}
928
929impl From<InvalidNumericMaxScaleError> for PlanError {
930    fn from(e: InvalidNumericMaxScaleError) -> PlanError {
931        PlanError::InvalidNumericMaxScale(e)
932    }
933}
934
935impl From<InvalidCharLengthError> for PlanError {
936    fn from(e: InvalidCharLengthError) -> PlanError {
937        PlanError::InvalidCharLength(e)
938    }
939}
940
941impl From<InvalidVarCharMaxLengthError> for PlanError {
942    fn from(e: InvalidVarCharMaxLengthError) -> PlanError {
943        PlanError::InvalidVarCharMaxLength(e)
944    }
945}
946
947impl From<InvalidTimestampPrecisionError> for PlanError {
948    fn from(e: InvalidTimestampPrecisionError) -> PlanError {
949        PlanError::InvalidTimestampPrecision(e)
950    }
951}
952
953impl From<anyhow::Error> for PlanError {
954    fn from(e: anyhow::Error) -> PlanError {
955        // WIP: Do we maybe want to keep the alternate selector for these?
956        sql_err!("{}", e.display_with_causes())
957    }
958}
959
960impl From<TryFromIntError> for PlanError {
961    fn from(e: TryFromIntError) -> PlanError {
962        sql_err!("{}", e.display_with_causes())
963    }
964}
965
966impl From<ParseIntError> for PlanError {
967    fn from(e: ParseIntError) -> PlanError {
968        sql_err!("{}", e.display_with_causes())
969    }
970}
971
972impl From<EvalError> for PlanError {
973    fn from(e: EvalError) -> PlanError {
974        sql_err!("{}", e.display_with_causes())
975    }
976}
977
978impl From<ParserError> for PlanError {
979    fn from(e: ParserError) -> PlanError {
980        PlanError::Parser(e)
981    }
982}
983
984impl From<ParserStatementError> for PlanError {
985    fn from(e: ParserStatementError) -> PlanError {
986        PlanError::ParserStatement(e)
987    }
988}
989
990impl From<PostgresError> for PlanError {
991    fn from(e: PostgresError) -> PlanError {
992        PlanError::PostgresConnectionErr { cause: Arc::new(e) }
993    }
994}
995
996impl From<MySqlError> for PlanError {
997    fn from(e: MySqlError) -> PlanError {
998        PlanError::MySqlConnectionErr { cause: Arc::new(e) }
999    }
1000}
1001
1002impl From<SqlServerError> for PlanError {
1003    fn from(e: SqlServerError) -> PlanError {
1004        PlanError::SqlServerConnectionErr { cause: Arc::new(e) }
1005    }
1006}
1007
1008impl From<VarError> for PlanError {
1009    fn from(e: VarError) -> Self {
1010        PlanError::VarError(e)
1011    }
1012}
1013
1014impl From<PgSourcePurificationError> for PlanError {
1015    fn from(e: PgSourcePurificationError) -> Self {
1016        PlanError::PgSourcePurification(e)
1017    }
1018}
1019
1020impl From<KafkaSourcePurificationError> for PlanError {
1021    fn from(e: KafkaSourcePurificationError) -> Self {
1022        PlanError::KafkaSourcePurification(e)
1023    }
1024}
1025
1026impl From<KafkaSinkPurificationError> for PlanError {
1027    fn from(e: KafkaSinkPurificationError) -> Self {
1028        PlanError::KafkaSinkPurification(e)
1029    }
1030}
1031
1032impl From<IcebergSinkPurificationError> for PlanError {
1033    fn from(e: IcebergSinkPurificationError) -> Self {
1034        PlanError::IcebergSinkPurification(e)
1035    }
1036}
1037
1038impl From<GluePurificationError> for PlanError {
1039    fn from(e: GluePurificationError) -> Self {
1040        PlanError::GluePurification(e)
1041    }
1042}
1043
1044impl From<CsrPurificationError> for PlanError {
1045    fn from(e: CsrPurificationError) -> Self {
1046        PlanError::CsrPurification(e)
1047    }
1048}
1049
1050impl From<LoadGeneratorSourcePurificationError> for PlanError {
1051    fn from(e: LoadGeneratorSourcePurificationError) -> Self {
1052        PlanError::LoadGeneratorSourcePurification(e)
1053    }
1054}
1055
1056impl From<MySqlSourcePurificationError> for PlanError {
1057    fn from(e: MySqlSourcePurificationError) -> Self {
1058        PlanError::MySqlSourcePurification(e)
1059    }
1060}
1061
1062impl From<SqlServerSourcePurificationError> for PlanError {
1063    fn from(e: SqlServerSourcePurificationError) -> Self {
1064        PlanError::SqlServerSourcePurificationError(e)
1065    }
1066}
1067
1068impl From<IdentError> for PlanError {
1069    fn from(e: IdentError) -> Self {
1070        PlanError::InvalidIdent(e)
1071    }
1072}
1073
1074impl From<ExternalReferenceResolutionError> for PlanError {
1075    fn from(e: ExternalReferenceResolutionError) -> Self {
1076        PlanError::SubsourceResolutionError(e)
1077    }
1078}
1079
1080struct ColumnDisplay<'a> {
1081    table: &'a Option<PartialItemName>,
1082    column: &'a ColumnName,
1083}
1084
1085impl<'a> fmt::Display for ColumnDisplay<'a> {
1086    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
1087        if let Some(table) = &self.table {
1088            format!("{}.{}", table.item, self.column).quoted().fmt(f)
1089        } else {
1090            self.column.quoted().fmt(f)
1091        }
1092    }
1093}
mz_sql/plan/error.rs

mz_sql/plan/
error.rs
