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