mz_adapter/
notice.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::fmt;
11use std::time::Duration;
12
13use chrono::{DateTime, Utc};
14use itertools::Itertools;
15use mz_controller::clusters::ClusterStatus;
16use mz_orchestrator::{OfflineReason, ServiceStatus};
17use mz_ore::str::{StrExt, separated};
18use mz_pgwire_common::{ErrorResponse, Severity};
19use mz_repr::adt::mz_acl_item::AclMode;
20use mz_repr::strconv;
21use mz_sql::ast::NoticeSeverity;
22use mz_sql::catalog::ErrorMessageObjectDescription;
23use mz_sql::plan::PlanNotice;
24use mz_sql::session::vars::IsolationLevel;
25use tokio_postgres::error::SqlState;
26
27use crate::TimestampExplanation;
28
29/// Notices that can occur in the adapter layer.
30///
31/// These are diagnostic warnings or informational messages that are not
32/// severe enough to warrant failing a query entirely.
33#[derive(Clone, Debug)]
34pub enum AdapterNotice {
35    DatabaseAlreadyExists {
36        name: String,
37    },
38    SchemaAlreadyExists {
39        name: String,
40    },
41    TableAlreadyExists {
42        name: String,
43    },
44    ObjectAlreadyExists {
45        name: String,
46        ty: &'static str,
47    },
48    DatabaseDoesNotExist {
49        name: String,
50    },
51    ClusterDoesNotExist {
52        name: String,
53    },
54    DefaultClusterDoesNotExist {
55        name: String,
56        kind: &'static str,
57        suggested_action: String,
58    },
59    NoResolvableSearchPathSchema {
60        search_path: Vec<String>,
61    },
62    ExistingTransactionInProgress,
63    ExplicitTransactionControlInImplicitTransaction,
64    UserRequested {
65        severity: NoticeSeverity,
66    },
67    ClusterReplicaStatusChanged {
68        cluster: String,
69        replica: String,
70        status: ClusterStatus,
71        time: DateTime<Utc>,
72    },
73    CascadeDroppedObject {
74        objects: Vec<String>,
75    },
76    DroppedActiveDatabase {
77        name: String,
78    },
79    DroppedActiveCluster {
80        name: String,
81    },
82    QueryTimestamp {
83        explanation: TimestampExplanation,
84    },
85    EqualSubscribeBounds {
86        bound: mz_repr::Timestamp,
87    },
88    QueryTrace {
89        trace_id: opentelemetry::trace::TraceId,
90    },
91    UnimplementedIsolationLevel {
92        isolation_level: String,
93    },
94    StrongSessionSerializable,
95    BadStartupSetting {
96        name: String,
97        reason: String,
98    },
99    RbacUserDisabled,
100    RoleMembershipAlreadyExists {
101        role_name: String,
102        member_name: String,
103    },
104    RoleMembershipDoesNotExists {
105        role_name: String,
106        member_name: String,
107    },
108    AutoRunOnCatalogServerCluster,
109    AlterIndexOwner {
110        name: String,
111    },
112    CannotRevoke {
113        object_description: ErrorMessageObjectDescription,
114    },
115    NonApplicablePrivilegeTypes {
116        non_applicable_privileges: AclMode,
117        object_description: ErrorMessageObjectDescription,
118    },
119    PlanNotice(PlanNotice),
120    UnknownSessionDatabase(String),
121    OptimizerNotice {
122        notice: String,
123        hint: String,
124    },
125    WebhookSourceCreated {
126        url: url::Url,
127    },
128    DroppedInUseIndex(DroppedInUseIndex),
129    DroppedIndexReadByInFlightStatements(InFlightIndexReaders),
130    PerReplicaLogRead {
131        log_names: Vec<String>,
132    },
133    VarDefaultUpdated {
134        role: Option<String>,
135        var_name: Option<String>,
136    },
137    /// An `ALTER SYSTEM` statement named a system parameter that
138    /// `environmentd` only reads at startup, so the running process keeps the
139    /// value it sampled at boot.
140    StartupOnlyVarUpdated {
141        var_name: String,
142    },
143    Welcome(String),
144    PlanInsights(String),
145    ExecutionTime(ExecutionTime),
146    IntrospectionClusterUsage,
147    AutoRouteIntrospectionQueriesUsage,
148    /// A cluster contains sources that run on only one replica, yet the
149    /// cluster has (or is about to have) more than one replica.
150    SingleReplicaSourcesOnMultiReplicaCluster {
151        cluster: String,
152        sources: Vec<String>,
153    },
154    /// An OIDC group maps to a reserved role name (mz_/pg_ prefix).
155    OidcGroupSyncReservedRole {
156        group: String,
157    },
158    /// OIDC group sync encountered an error (fail-open mode).
159    OidcGroupSyncError {
160        message: String,
161    },
162}
163
164impl AdapterNotice {
165    pub fn into_response(self) -> ErrorResponse {
166        ErrorResponse {
167            severity: self.severity(),
168            code: self.code(),
169            message: self.to_string(),
170            detail: self.detail(),
171            hint: self.hint(),
172            position: None,
173        }
174    }
175
176    /// Returns the severity for a notice.
177    pub fn severity(&self) -> Severity {
178        match self {
179            AdapterNotice::DatabaseAlreadyExists { .. } => Severity::Notice,
180            AdapterNotice::SchemaAlreadyExists { .. } => Severity::Notice,
181            AdapterNotice::TableAlreadyExists { .. } => Severity::Notice,
182            AdapterNotice::ObjectAlreadyExists { .. } => Severity::Notice,
183            AdapterNotice::DatabaseDoesNotExist { .. } => Severity::Notice,
184            AdapterNotice::ClusterDoesNotExist { .. } => Severity::Notice,
185            AdapterNotice::DefaultClusterDoesNotExist { .. } => Severity::Notice,
186            AdapterNotice::NoResolvableSearchPathSchema { .. } => Severity::Notice,
187            AdapterNotice::ExistingTransactionInProgress => Severity::Warning,
188            AdapterNotice::ExplicitTransactionControlInImplicitTransaction => Severity::Warning,
189            AdapterNotice::UserRequested { severity } => match severity {
190                NoticeSeverity::Debug => Severity::Debug,
191                NoticeSeverity::Info => Severity::Info,
192                NoticeSeverity::Log => Severity::Log,
193                NoticeSeverity::Notice => Severity::Notice,
194                NoticeSeverity::Warning => Severity::Warning,
195            },
196            AdapterNotice::ClusterReplicaStatusChanged { .. } => Severity::Notice,
197            AdapterNotice::CascadeDroppedObject { .. } => Severity::Notice,
198            AdapterNotice::DroppedActiveDatabase { .. } => Severity::Notice,
199            AdapterNotice::DroppedActiveCluster { .. } => Severity::Notice,
200            AdapterNotice::QueryTimestamp { .. } => Severity::Notice,
201            AdapterNotice::EqualSubscribeBounds { .. } => Severity::Notice,
202            AdapterNotice::QueryTrace { .. } => Severity::Notice,
203            AdapterNotice::UnimplementedIsolationLevel { .. } => Severity::Notice,
204            AdapterNotice::StrongSessionSerializable => Severity::Notice,
205            AdapterNotice::BadStartupSetting { .. } => Severity::Notice,
206            AdapterNotice::RbacUserDisabled => Severity::Notice,
207            AdapterNotice::RoleMembershipAlreadyExists { .. } => Severity::Notice,
208            AdapterNotice::RoleMembershipDoesNotExists { .. } => Severity::Warning,
209            AdapterNotice::AutoRunOnCatalogServerCluster => Severity::Debug,
210            AdapterNotice::AlterIndexOwner { .. } => Severity::Warning,
211            AdapterNotice::CannotRevoke { .. } => Severity::Warning,
212            AdapterNotice::NonApplicablePrivilegeTypes { .. } => Severity::Notice,
213            AdapterNotice::PlanNotice(notice) => match notice {
214                PlanNotice::ObjectDoesNotExist { .. } => Severity::Notice,
215                PlanNotice::ColumnAlreadyExists { .. } => Severity::Notice,
216                PlanNotice::UpsertSinkKeyNotEnforced { .. } => Severity::Warning,
217                PlanNotice::ReplicaDiskOptionDeprecated { .. } => Severity::Notice,
218            },
219            AdapterNotice::UnknownSessionDatabase(_) => Severity::Notice,
220            AdapterNotice::OptimizerNotice { .. } => Severity::Notice,
221            AdapterNotice::WebhookSourceCreated { .. } => Severity::Notice,
222            AdapterNotice::DroppedInUseIndex { .. } => Severity::Warning,
223            AdapterNotice::DroppedIndexReadByInFlightStatements(_) => Severity::Notice,
224            AdapterNotice::PerReplicaLogRead { .. } => Severity::Notice,
225            AdapterNotice::VarDefaultUpdated { .. } => Severity::Notice,
226            AdapterNotice::StartupOnlyVarUpdated { .. } => Severity::Warning,
227            AdapterNotice::Welcome(_) => Severity::Notice,
228            AdapterNotice::PlanInsights(_) => Severity::Notice,
229            AdapterNotice::ExecutionTime(_) => Severity::Notice,
230            AdapterNotice::IntrospectionClusterUsage => Severity::Warning,
231            AdapterNotice::AutoRouteIntrospectionQueriesUsage => Severity::Warning,
232            AdapterNotice::SingleReplicaSourcesOnMultiReplicaCluster { .. } => Severity::Warning,
233            AdapterNotice::OidcGroupSyncReservedRole { .. } => Severity::Warning,
234            AdapterNotice::OidcGroupSyncError { .. } => Severity::Warning,
235        }
236    }
237
238    /// Reports additional details about the notice, if any are available.
239    pub fn detail(&self) -> Option<String> {
240        match self {
241            AdapterNotice::PlanNotice(notice) => notice.detail(),
242            AdapterNotice::ExecutionTime(time) => Some(
243                serde_json::json!({
244                    "kind": time.kind.name(),
245                    "duration_us": u64::try_from(time.elapsed.as_micros()).unwrap_or(u64::MAX),
246                })
247                .to_string(),
248            ),
249            AdapterNotice::DroppedInUseIndex(..) => Some(
250                "The index no longer appears in the catalog, but it is still maintained, using \
251                 memory and CPU, until those objects are dropped or recreated, or replanned at \
252                 the next restart. The drop was allowed because enable_drop_in_use_index is set."
253                    .into(),
254            ),
255            AdapterNotice::SingleReplicaSourcesOnMultiReplicaCluster { .. } => Some(
256                "Adding replicas to the cluster does not make these sources more fault tolerant \
257                 and does not increase their ingestion throughput."
258                    .into(),
259            ),
260            AdapterNotice::QueryTimestamp { explanation } => Some(format!("\n{explanation}")),
261            AdapterNotice::CascadeDroppedObject { objects } => Some(
262                objects
263                    .iter()
264                    .map(|obj_info| format!("drop cascades to {}", obj_info))
265                    .join("\n"),
266            ),
267            _ => None,
268        }
269    }
270
271    /// Reports a hint for the user about how the notice could be addressed.
272    pub fn hint(&self) -> Option<String> {
273        match self {
274            AdapterNotice::DatabaseDoesNotExist { name: _ } => Some("Create the database with CREATE DATABASE or pick an extant database with SET DATABASE = name. List available databases with SHOW DATABASES.".into()),
275            AdapterNotice::ClusterDoesNotExist { name: _ } => Some("Create the cluster with CREATE CLUSTER or pick an extant cluster with SET CLUSTER = name. List available clusters with SHOW CLUSTERS.".into()),
276            AdapterNotice::DefaultClusterDoesNotExist {
277                name: _,
278                kind: _,
279                suggested_action,
280            } => Some(suggested_action.clone()),
281            AdapterNotice::NoResolvableSearchPathSchema { search_path: _ } => Some("Create a schema with CREATE SCHEMA or pick an extant schema with SET SCHEMA = name. List available schemas with SHOW SCHEMAS.".into()),
282            AdapterNotice::DroppedActiveDatabase { name: _ } => Some("Choose a new active database by executing SET DATABASE = <name>.".into()),
283            AdapterNotice::DroppedActiveCluster { name: _ } => Some("Choose a new active cluster by executing SET CLUSTER = <name>.".into()),
284            AdapterNotice::ClusterReplicaStatusChanged { status, .. } => {
285                match status {
286                    ServiceStatus::Offline(None)
287                    | ServiceStatus::Offline(Some(OfflineReason::Initializing)) => Some("The cluster replica may be restarting or going offline.".into()),
288                    ServiceStatus::Offline(Some(OfflineReason::OomKilled)) => Some("The cluster replica may have run out of memory and been killed.".into()),
289                    ServiceStatus::Online => None,
290                }
291            },
292            AdapterNotice::RbacUserDisabled => Some("To enable RBAC globally run `ALTER SYSTEM SET enable_rbac_checks TO TRUE` as a superuser. TO enable RBAC for just this session run `SET enable_session_rbac_checks TO TRUE`.".into()),
293            AdapterNotice::AlterIndexOwner {name: _} => Some("Change the ownership of the index's relation, instead.".into()),
294            AdapterNotice::UnknownSessionDatabase(_) => Some(
295                "Create the database with CREATE DATABASE \
296                 or pick an extant database with SET DATABASE = name. \
297                 List available databases with SHOW DATABASES."
298                    .into(),
299            ),
300            AdapterNotice::OptimizerNotice { notice: _, hint } => Some(hint.clone()),
301            AdapterNotice::DroppedInUseIndex(..) => Some("Drop and recreate those objects to free the index's resources.".into()),
302            AdapterNotice::IntrospectionClusterUsage => Some("Use the new name instead.".into()),
303            AdapterNotice::AutoRouteIntrospectionQueriesUsage => Some("Use the new name instead.".into()),
304            AdapterNotice::SingleReplicaSourcesOnMultiReplicaCluster { .. } => Some(
305                "To achieve fault tolerance for other objects in the cluster, consider moving \
306                 these sources to a separate cluster with a single replica."
307                    .into(),
308            ),
309            _ => None
310        }
311    }
312
313    /// Reports the error code.
314    pub fn code(&self) -> SqlState {
315        match self {
316            AdapterNotice::DatabaseAlreadyExists { .. } => SqlState::DUPLICATE_DATABASE,
317            AdapterNotice::SchemaAlreadyExists { .. } => SqlState::DUPLICATE_SCHEMA,
318            AdapterNotice::TableAlreadyExists { .. } => SqlState::DUPLICATE_TABLE,
319            AdapterNotice::ObjectAlreadyExists { .. } => SqlState::DUPLICATE_OBJECT,
320            AdapterNotice::DatabaseDoesNotExist { .. } => SqlState::from_code("MZ006"),
321            AdapterNotice::ClusterDoesNotExist { .. } => SqlState::from_code("MZ007"),
322            AdapterNotice::NoResolvableSearchPathSchema { .. } => SqlState::from_code("MZ008"),
323            AdapterNotice::ExistingTransactionInProgress => SqlState::ACTIVE_SQL_TRANSACTION,
324            AdapterNotice::ExplicitTransactionControlInImplicitTransaction => {
325                SqlState::NO_ACTIVE_SQL_TRANSACTION
326            }
327            AdapterNotice::UserRequested { severity } => match severity {
328                NoticeSeverity::Warning => SqlState::WARNING,
329                _ => SqlState::SUCCESSFUL_COMPLETION,
330            },
331            AdapterNotice::ClusterReplicaStatusChanged { .. } => SqlState::SUCCESSFUL_COMPLETION,
332            AdapterNotice::CascadeDroppedObject { .. } => SqlState::SUCCESSFUL_COMPLETION,
333            AdapterNotice::DroppedActiveDatabase { .. } => SqlState::from_code("MZ002"),
334            AdapterNotice::DroppedActiveCluster { .. } => SqlState::from_code("MZ003"),
335            AdapterNotice::QueryTimestamp { .. } => SqlState::SUCCESSFUL_COMPLETION,
336            AdapterNotice::EqualSubscribeBounds { .. } => SqlState::SUCCESSFUL_COMPLETION,
337            AdapterNotice::QueryTrace { .. } => SqlState::SUCCESSFUL_COMPLETION,
338            AdapterNotice::UnimplementedIsolationLevel { .. } => SqlState::SUCCESSFUL_COMPLETION,
339            AdapterNotice::StrongSessionSerializable => SqlState::SUCCESSFUL_COMPLETION,
340            AdapterNotice::BadStartupSetting { .. } => SqlState::SUCCESSFUL_COMPLETION,
341            AdapterNotice::RbacUserDisabled => SqlState::SUCCESSFUL_COMPLETION,
342            AdapterNotice::RoleMembershipAlreadyExists { .. } => SqlState::SUCCESSFUL_COMPLETION,
343            AdapterNotice::RoleMembershipDoesNotExists { .. } => SqlState::WARNING,
344            AdapterNotice::AutoRunOnCatalogServerCluster => SqlState::SUCCESSFUL_COMPLETION,
345            AdapterNotice::AlterIndexOwner { .. } => SqlState::WARNING,
346            AdapterNotice::CannotRevoke { .. } => SqlState::WARNING_PRIVILEGE_NOT_REVOKED,
347            AdapterNotice::NonApplicablePrivilegeTypes { .. } => SqlState::SUCCESSFUL_COMPLETION,
348            AdapterNotice::PlanNotice(plan) => match plan {
349                PlanNotice::ObjectDoesNotExist { .. } => SqlState::UNDEFINED_OBJECT,
350                PlanNotice::ColumnAlreadyExists { .. } => SqlState::DUPLICATE_COLUMN,
351                PlanNotice::UpsertSinkKeyNotEnforced { .. } => SqlState::WARNING,
352                PlanNotice::ReplicaDiskOptionDeprecated { .. } => {
353                    SqlState::WARNING_DEPRECATED_FEATURE
354                }
355            },
356            AdapterNotice::UnknownSessionDatabase(_) => SqlState::from_code("MZ004"),
357            AdapterNotice::DefaultClusterDoesNotExist { .. } => SqlState::from_code("MZ005"),
358            AdapterNotice::OptimizerNotice { .. } => SqlState::SUCCESSFUL_COMPLETION,
359            AdapterNotice::DroppedInUseIndex { .. } => SqlState::WARNING,
360            AdapterNotice::DroppedIndexReadByInFlightStatements(_) => {
361                SqlState::SUCCESSFUL_COMPLETION
362            }
363            AdapterNotice::WebhookSourceCreated { .. } => SqlState::SUCCESSFUL_COMPLETION,
364            AdapterNotice::PerReplicaLogRead { .. } => SqlState::SUCCESSFUL_COMPLETION,
365            AdapterNotice::VarDefaultUpdated { .. } => SqlState::SUCCESSFUL_COMPLETION,
366            AdapterNotice::StartupOnlyVarUpdated { .. } => SqlState::WARNING,
367            AdapterNotice::Welcome(_) => SqlState::SUCCESSFUL_COMPLETION,
368            AdapterNotice::PlanInsights(_) => SqlState::from_code("MZ001"),
369            AdapterNotice::ExecutionTime(_) => SqlState::from_code("MZ012"),
370            AdapterNotice::IntrospectionClusterUsage => SqlState::WARNING,
371            AdapterNotice::AutoRouteIntrospectionQueriesUsage => SqlState::WARNING,
372            AdapterNotice::SingleReplicaSourcesOnMultiReplicaCluster { .. } => SqlState::WARNING,
373            AdapterNotice::OidcGroupSyncReservedRole { .. } => SqlState::WARNING,
374            AdapterNotice::OidcGroupSyncError { .. } => SqlState::WARNING,
375        }
376    }
377}
378
379impl fmt::Display for AdapterNotice {
380    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
381        match self {
382            AdapterNotice::DatabaseAlreadyExists { name } => {
383                write!(f, "database {} already exists, skipping", name.quoted())
384            }
385            AdapterNotice::SchemaAlreadyExists { name } => {
386                write!(f, "schema {} already exists, skipping", name.quoted())
387            }
388            AdapterNotice::TableAlreadyExists { name } => {
389                write!(f, "table {} already exists, skipping", name.quoted())
390            }
391            AdapterNotice::ObjectAlreadyExists { name, ty } => {
392                write!(f, "{} {} already exists, skipping", ty, name.quoted())
393            }
394            AdapterNotice::DatabaseDoesNotExist { name } => {
395                write!(f, "database {} does not exist", name.quoted())
396            }
397            AdapterNotice::CascadeDroppedObject { objects } => {
398                write!(f, "drop cascades to {} other objects", objects.len())
399            }
400            AdapterNotice::ClusterDoesNotExist { name } => {
401                write!(f, "cluster {} does not exist", name.quoted())
402            }
403            AdapterNotice::DefaultClusterDoesNotExist { kind, name, .. } => {
404                write!(f, "{kind} default cluster {} does not exist", name.quoted())
405            }
406            AdapterNotice::NoResolvableSearchPathSchema { search_path } => {
407                write!(
408                    f,
409                    "no schema on the search path exists: {}",
410                    search_path.join(", ")
411                )
412            }
413            AdapterNotice::ExistingTransactionInProgress => {
414                write!(f, "there is already a transaction in progress")
415            }
416            AdapterNotice::ExplicitTransactionControlInImplicitTransaction => {
417                write!(f, "there is no transaction in progress")
418            }
419            AdapterNotice::UserRequested { severity } => {
420                write!(f, "raised a test {}", severity.to_string().to_lowercase())
421            }
422            AdapterNotice::ClusterReplicaStatusChanged {
423                cluster,
424                replica,
425                status,
426                time,
427            } => {
428                let mut time_buf = String::new();
429                strconv::format_timestamptz(&mut time_buf, time);
430                write!(
431                    f,
432                    "cluster replica {}.{} changed status to {} at {}",
433                    cluster,
434                    replica,
435                    status.as_kebab_case_str().quoted(),
436                    time_buf,
437                )?;
438                Ok(())
439            }
440            AdapterNotice::DroppedActiveDatabase { name } => {
441                write!(f, "active database {} has been dropped", name.quoted())
442            }
443            AdapterNotice::DroppedActiveCluster { name } => {
444                write!(f, "active cluster {} has been dropped", name.quoted())
445            }
446            AdapterNotice::QueryTimestamp { .. } => write!(f, "EXPLAIN TIMESTAMP for query"),
447            AdapterNotice::EqualSubscribeBounds { bound } => {
448                write!(
449                    f,
450                    "subscribe as of {bound} (inclusive) up to the same bound {bound} (exclusive) is guaranteed to be empty"
451                )
452            }
453            AdapterNotice::QueryTrace { trace_id } => {
454                write!(f, "trace id: {}", trace_id)
455            }
456            AdapterNotice::UnimplementedIsolationLevel { isolation_level } => {
457                write!(
458                    f,
459                    "transaction isolation level {isolation_level} is unimplemented, the session will be upgraded to {}",
460                    IsolationLevel::Serializable
461                )
462            }
463            AdapterNotice::StrongSessionSerializable => {
464                write!(
465                    f,
466                    "The Strong Session Serializable isolation level may exhibit consistency violations when reading from catalog objects",
467                )
468            }
469            AdapterNotice::BadStartupSetting { name, reason } => {
470                write!(f, "startup setting {name} not set: {reason}")
471            }
472            AdapterNotice::RbacUserDisabled => {
473                write!(
474                    f,
475                    "RBAC is disabled so no role attributes or object ownership will be considered \
476                    when executing statements"
477                )
478            }
479            AdapterNotice::RoleMembershipAlreadyExists {
480                role_name,
481                member_name,
482            } => write!(
483                f,
484                "role \"{member_name}\" is already a member of role \"{role_name}\""
485            ),
486            AdapterNotice::RoleMembershipDoesNotExists {
487                role_name,
488                member_name,
489            } => write!(
490                f,
491                "role \"{member_name}\" is not a member of role \"{role_name}\""
492            ),
493            AdapterNotice::AutoRunOnCatalogServerCluster => write!(
494                f,
495                "query was automatically run on the \"mz_catalog_server\" cluster"
496            ),
497            AdapterNotice::AlterIndexOwner { name } => {
498                write!(f, "cannot change owner of {}", name.quoted())
499            }
500            AdapterNotice::CannotRevoke { object_description } => {
501                write!(f, "no privileges could be revoked for {object_description}")
502            }
503            AdapterNotice::NonApplicablePrivilegeTypes {
504                non_applicable_privileges,
505                object_description,
506            } => {
507                write!(
508                    f,
509                    "non-applicable privilege types {} for {}",
510                    non_applicable_privileges.to_error_string(),
511                    object_description,
512                )
513            }
514            AdapterNotice::PlanNotice(plan) => plan.fmt(f),
515            AdapterNotice::UnknownSessionDatabase(name) => {
516                write!(f, "session database {} does not exist", name.quoted())
517            }
518            AdapterNotice::OptimizerNotice { notice, hint: _ } => notice.fmt(f),
519            AdapterNotice::WebhookSourceCreated { url } => {
520                write!(f, "URL to POST data is '{url}'")
521            }
522            AdapterNotice::DroppedInUseIndex(DroppedInUseIndex {
523                index_name,
524                dependant_objects,
525            }) => {
526                write!(
527                    f,
528                    "index {} was dropped, but these objects still read from it: {}",
529                    index_name.quoted(),
530                    separated(", ", dependant_objects)
531                )
532            }
533            AdapterNotice::DroppedIndexReadByInFlightStatements(readers) => {
534                write!(
535                    f,
536                    "index {} is still read by transient dataflows for {}. Those dataflows end when their statements finish, and the index is maintained until then.",
537                    readers.index_name.quoted(),
538                    readers.describe_statements()
539                )
540            }
541            AdapterNotice::PerReplicaLogRead { log_names } => {
542                write!(
543                    f,
544                    "Queried introspection relations: {}. Unlike other objects in Materialize, results from querying these objects depend on the current values of the `cluster` and `cluster_replica` session variables.",
545                    log_names.join(", ")
546                )
547            }
548            AdapterNotice::VarDefaultUpdated { role, var_name } => {
549                let vars = match var_name {
550                    Some(name) => format!("variable {} was", name.quoted()),
551                    None => "variables were".to_string(),
552                };
553                let target = match role {
554                    Some(role_name) => role_name.quoted().to_string(),
555                    None => "the system".to_string(),
556                };
557                write!(
558                    f,
559                    "{vars} updated for {target}, this will have no effect on the current session"
560                )
561            }
562            AdapterNotice::StartupOnlyVarUpdated { var_name } => write!(
563                f,
564                "changes to {} only take effect when environmentd restarts",
565                var_name.quoted()
566            ),
567            AdapterNotice::Welcome(message) => message.fmt(f),
568            AdapterNotice::PlanInsights(message) => message.fmt(f),
569            AdapterNotice::ExecutionTime(time) => {
570                write!(
571                    f,
572                    "execution time: {:.3} ms",
573                    time.elapsed.as_secs_f64() * 1000.0
574                )?;
575                if matches!(time.kind, ExecutionTimeKind::Staged) {
576                    f.write_str(" (staged)")?;
577                }
578                Ok(())
579            }
580            AdapterNotice::IntrospectionClusterUsage => write!(
581                f,
582                "The mz_introspection cluster has been renamed to mz_catalog_server."
583            ),
584            AdapterNotice::AutoRouteIntrospectionQueriesUsage => write!(
585                f,
586                "The auto_route_introspection_queries variable has been renamed to auto_route_catalog_queries."
587            ),
588            AdapterNotice::SingleReplicaSourcesOnMultiReplicaCluster { cluster, sources } => {
589                // Keep the notice readable on clusters with many sources.
590                const MAX_LISTED_SOURCES: usize = 3;
591                write!(
592                    f,
593                    "cluster {} has more than one replica, but the following sources in it always \
594                     run on only the first replica: {}",
595                    cluster.quoted(),
596                    separated(
597                        ", ",
598                        sources.iter().take(MAX_LISTED_SOURCES).map(|n| n.quoted())
599                    ),
600                )?;
601                if sources.len() > MAX_LISTED_SOURCES {
602                    write!(f, ", and {} more", sources.len() - MAX_LISTED_SOURCES)?;
603                }
604                Ok(())
605            }
606            AdapterNotice::OidcGroupSyncReservedRole { group } => {
607                write!(
608                    f,
609                    "OIDC group \"{}\" maps to a reserved role name, skipping",
610                    group
611                )
612            }
613            AdapterNotice::OidcGroupSyncError { message } => {
614                write!(f, "OIDC group-to-role sync failed: {}", message)
615            }
616        }
617    }
618}
619
620#[derive(Clone, Debug)]
621pub struct DroppedInUseIndex {
622    pub index_name: String,
623    pub dependant_objects: Vec<String>,
624}
625
626/// One-shot dataflows that read from a dropped index, by the kind of statement that owns them.
627#[derive(Clone, Debug, Default)]
628pub struct InFlightIndexReaders {
629    pub index_name: String,
630    pub subscribes: usize,
631    pub copy_tos: usize,
632    pub selects: usize,
633    /// Dataflows the system installed itself, such as introspection subscribes.
634    pub system: usize,
635}
636
637impl InFlightIndexReaders {
638    fn describe_statements(&self) -> String {
639        fn part(count: usize, what: &str) -> Option<String> {
640            let plural = if count == 1 { "" } else { "s" };
641            (count > 0).then(|| format!("{count} in-progress {what}{plural}"))
642        }
643        [
644            part(self.subscribes, "SUBSCRIBE statement"),
645            part(self.copy_tos, "COPY TO statement"),
646            part(self.selects, "SELECT statement"),
647            part(self.system, "system dataflow"),
648        ]
649        .into_iter()
650        .flatten()
651        .join(", ")
652    }
653}
654
655impl From<PlanNotice> for AdapterNotice {
656    fn from(notice: PlanNotice) -> AdapterNotice {
657        AdapterNotice::PlanNotice(notice)
658    }
659}
660
661/// Time from bound-statement execution to its first available result.
662///
663/// Parsing, describing, and binding are excluded, except binding within SQL `EXECUTE`.
664/// Reads and writes use the same boundary. Subsequent row delivery and implicit commit
665/// are excluded. Work performed within execution, including an explicit `COMMIT`, is
666/// included.
667#[derive(Clone, Debug)]
668pub struct ExecutionTime {
669    pub(crate) kind: ExecutionTimeKind,
670    pub(crate) elapsed: Duration,
671}
672
673impl ExecutionTime {
674    /// The time of a statement that cannot return rows.
675    pub fn completed(elapsed: Duration) -> Self {
676        Self {
677            kind: ExecutionTimeKind::Completed,
678            elapsed,
679        }
680    }
681}
682
683/// Where the interval of an [`ExecutionTime`] ends.
684#[derive(Clone, Copy, Debug)]
685pub(crate) enum ExecutionTimeKind {
686    /// A statement that returns rows has its first row ready. Delivering the rows is excluded.
687    FirstRow,
688    /// A statement that could return rows finished without any.
689    EmptyResult,
690    /// A statement that cannot return rows, such as `SET`, finished executing.
691    Completed,
692    /// The transaction still holds work for a later commit, which the time excludes.
693    Staged,
694}
695
696impl ExecutionTimeKind {
697    fn name(self) -> &'static str {
698        match self {
699            Self::FirstRow => "first_row",
700            Self::EmptyResult => "empty_result",
701            Self::Completed => "completed",
702            Self::Staged => "staged",
703        }
704    }
705}
mz_adapter/notice.rs

mz_adapter/
notice.rs
