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

mz_sql/plan/
error.rs
