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

mz_sql/plan/
error.rs
