mz_adapter/coord/sequencer/inner/
create_materialized_view.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 anyhow::anyhow;
11use differential_dataflow::lattice::Lattice;
12use maplit::btreemap;
13use maplit::btreeset;
14use mz_adapter_types::compaction::CompactionWindow;
15use mz_catalog::memory::objects::{CatalogItem, MaterializedView};
16use mz_expr::{CollectionPlan, ResultSpec};
17use mz_ore::collections::CollectionExt;
18use mz_ore::instrument;
19use mz_ore::soft_panic_or_log;
20use mz_repr::explain::{ExprHumanizerExt, TransientItem};
21use mz_repr::optimize::OptimizerFeatures;
22use mz_repr::optimize::OverrideFrom;
23use mz_repr::refresh_schedule::RefreshSchedule;
24use mz_repr::{CatalogItemId, Datum, Row};
25use mz_sql::ast::ExplainStage;
26use mz_sql::catalog::CatalogError;
27use mz_sql::names::ResolvedIds;
28use mz_sql::plan;
29use mz_sql::session::metadata::SessionMetadata;
30use mz_sql_parser::ast;
31use mz_sql_parser::ast::display::AstDisplay;
32use mz_storage_client::controller::{CollectionDescription, DataSource};
33use std::collections::BTreeMap;
34use timely::progress::Antichain;
35use tracing::Span;
36
37use crate::ReadHolds;
38use crate::command::ExecuteResponse;
39use crate::coord::sequencer::inner::return_if_err;
40use crate::coord::{
41    Coordinator, CreateMaterializedViewExplain, CreateMaterializedViewFinish,
42    CreateMaterializedViewOptimize, CreateMaterializedViewStage, ExplainContext,
43    ExplainPlanContext, Message, PlanValidity, StageResult, Staged,
44};
45use crate::error::AdapterError;
46use crate::explain::explain_dataflow;
47use crate::explain::explain_plan;
48use crate::explain::optimizer_trace::OptimizerTrace;
49use crate::optimize::dataflows::dataflow_import_id_bundle;
50use crate::optimize::{self, Optimize};
51use crate::session::Session;
52use crate::util::ResultExt;
53use crate::{AdapterNotice, CollectionIdBundle, ExecuteContext, TimestampProvider, catalog};
54
55impl Staged for CreateMaterializedViewStage {
56    type Ctx = ExecuteContext;
57
58    fn validity(&mut self) -> &mut PlanValidity {
59        match self {
60            Self::Optimize(stage) => &mut stage.validity,
61            Self::Finish(stage) => &mut stage.validity,
62            Self::Explain(stage) => &mut stage.validity,
63        }
64    }
65
66    async fn stage(
67        self,
68        coord: &mut Coordinator,
69        ctx: &mut ExecuteContext,
70    ) -> Result<StageResult<Box<Self>>, AdapterError> {
71        match self {
72            CreateMaterializedViewStage::Optimize(stage) => {
73                coord.create_materialized_view_optimize(stage).await
74            }
75            CreateMaterializedViewStage::Finish(stage) => {
76                coord.create_materialized_view_finish(ctx, stage).await
77            }
78            CreateMaterializedViewStage::Explain(stage) => {
79                coord
80                    .create_materialized_view_explain(ctx.session(), stage)
81                    .await
82            }
83        }
84    }
85
86    fn message(self, ctx: ExecuteContext, span: Span) -> Message {
87        Message::CreateMaterializedViewStageReady {
88            ctx,
89            span,
90            stage: self,
91        }
92    }
93
94    fn cancel_enabled(&self) -> bool {
95        true
96    }
97}
98
99impl Coordinator {
100    #[instrument]
101    pub(crate) async fn sequence_create_materialized_view(
102        &mut self,
103        ctx: ExecuteContext,
104        plan: plan::CreateMaterializedViewPlan,
105        resolved_ids: ResolvedIds,
106    ) {
107        let stage = return_if_err!(
108            self.create_materialized_view_validate(
109                ctx.session(),
110                plan,
111                resolved_ids,
112                ExplainContext::None
113            ),
114            ctx
115        );
116        self.sequence_staged(ctx, Span::current(), stage).await;
117    }
118
119    #[instrument]
120    pub(crate) async fn explain_create_materialized_view(
121        &mut self,
122        ctx: ExecuteContext,
123        plan::ExplainPlanPlan {
124            stage,
125            format,
126            config,
127            explainee,
128        }: plan::ExplainPlanPlan,
129    ) {
130        let plan::Explainee::Statement(stmt) = explainee else {
131            // This is currently asserted in the `sequence_explain_plan` code that
132            // calls this method.
133            unreachable!()
134        };
135        let plan::ExplaineeStatement::CreateMaterializedView { broken, plan } = stmt else {
136            // This is currently asserted in the `sequence_explain_plan` code that
137            // calls this method.
138            unreachable!()
139        };
140
141        // Create an OptimizerTrace instance to collect plans emitted when
142        // executing the optimizer pipeline.
143        let optimizer_trace = OptimizerTrace::new(stage.paths());
144
145        // Not used in the EXPLAIN path so it's OK to generate a dummy value.
146        let resolved_ids = ResolvedIds::empty();
147
148        let explain_ctx = ExplainContext::Plan(ExplainPlanContext {
149            broken,
150            config,
151            format,
152            stage,
153            replan: None,
154            desc: None,
155            optimizer_trace,
156        });
157        let stage = return_if_err!(
158            self.create_materialized_view_validate(ctx.session(), plan, resolved_ids, explain_ctx),
159            ctx
160        );
161        self.sequence_staged(ctx, Span::current(), stage).await;
162    }
163
164    #[instrument]
165    pub(crate) async fn explain_replan_materialized_view(
166        &mut self,
167        ctx: ExecuteContext,
168        plan::ExplainPlanPlan {
169            stage,
170            format,
171            config,
172            explainee,
173        }: plan::ExplainPlanPlan,
174    ) {
175        let plan::Explainee::ReplanMaterializedView(id) = explainee else {
176            unreachable!() // Asserted in `sequence_explain_plan`.
177        };
178        let CatalogItem::MaterializedView(item) = self.catalog().get_entry(&id).item() else {
179            unreachable!() // Asserted in `plan_explain_plan`.
180        };
181        let gid = item.global_id();
182
183        let create_sql = item.create_sql.clone();
184        let plan_result = self
185            .catalog_mut()
186            .deserialize_plan_with_enable_for_item_parsing(&create_sql, true);
187        let (plan, resolved_ids) = return_if_err!(plan_result, ctx);
188
189        let plan::Plan::CreateMaterializedView(plan) = plan else {
190            unreachable!() // We are parsing the `create_sql` of a `MaterializedView` item.
191        };
192
193        // It is safe to assume that query optimization will always succeed, so
194        // for now we statically assume `broken = false`.
195        let broken = false;
196
197        // Create an OptimizerTrace instance to collect plans emitted when
198        // executing the optimizer pipeline.
199        let optimizer_trace = OptimizerTrace::new(stage.paths());
200
201        let explain_ctx = ExplainContext::Plan(ExplainPlanContext {
202            broken,
203            config,
204            format,
205            stage,
206            replan: Some(gid),
207            desc: None,
208            optimizer_trace,
209        });
210        let stage = return_if_err!(
211            self.create_materialized_view_validate(ctx.session(), plan, resolved_ids, explain_ctx,),
212            ctx
213        );
214        self.sequence_staged(ctx, Span::current(), stage).await;
215    }
216
217    #[instrument]
218    pub(super) fn explain_materialized_view(
219        &mut self,
220        ctx: &ExecuteContext,
221        plan::ExplainPlanPlan {
222            stage,
223            format,
224            config,
225            explainee,
226        }: plan::ExplainPlanPlan,
227    ) -> Result<ExecuteResponse, AdapterError> {
228        let plan::Explainee::MaterializedView(id) = explainee else {
229            unreachable!() // Asserted in `sequence_explain_plan`.
230        };
231        let CatalogItem::MaterializedView(view) = self.catalog().get_entry(&id).item() else {
232            unreachable!() // Asserted in `plan_explain_plan`.
233        };
234        let gid = view.global_id();
235
236        let Some(dataflow_metainfo) = self.catalog().try_get_dataflow_metainfo(&gid) else {
237            if !id.is_system() {
238                tracing::error!(
239                    "cannot find dataflow metainformation for materialized view {id} in catalog"
240                );
241            }
242            coord_bail!(
243                "cannot find dataflow metainformation for materialized view {id} in catalog"
244            );
245        };
246
247        let target_cluster = self.catalog().get_cluster(view.cluster_id);
248
249        let features = OptimizerFeatures::from(self.catalog().system_config())
250            .override_from(&target_cluster.config.features())
251            .override_from(&config.features);
252
253        let cardinality_stats = BTreeMap::new();
254
255        let explain = match stage {
256            ExplainStage::RawPlan => explain_plan(
257                view.raw_expr.as_ref().clone(),
258                format,
259                &config,
260                &features,
261                &self.catalog().for_session(ctx.session()),
262                cardinality_stats,
263                Some(target_cluster.name.as_str()),
264            )?,
265            ExplainStage::LocalPlan => explain_plan(
266                view.optimized_expr.as_inner().clone(),
267                format,
268                &config,
269                &features,
270                &self.catalog().for_session(ctx.session()),
271                cardinality_stats,
272                Some(target_cluster.name.as_str()),
273            )?,
274            ExplainStage::GlobalPlan => {
275                let Some(plan) = self.catalog().try_get_optimized_plan(&gid).cloned() else {
276                    tracing::error!("cannot find {stage} for materialized view {id} in catalog");
277                    coord_bail!("cannot find {stage} for materialized view in catalog");
278                };
279                explain_dataflow(
280                    plan,
281                    format,
282                    &config,
283                    &features,
284                    &self.catalog().for_session(ctx.session()),
285                    cardinality_stats,
286                    Some(target_cluster.name.as_str()),
287                    dataflow_metainfo,
288                )?
289            }
290            ExplainStage::PhysicalPlan => {
291                let Some(plan) = self.catalog().try_get_physical_plan(&gid).cloned() else {
292                    tracing::error!("cannot find {stage} for materialized view {id} in catalog",);
293                    coord_bail!("cannot find {stage} for materialized view in catalog");
294                };
295                explain_dataflow(
296                    plan,
297                    format,
298                    &config,
299                    &features,
300                    &self.catalog().for_session(ctx.session()),
301                    cardinality_stats,
302                    Some(target_cluster.name.as_str()),
303                    dataflow_metainfo,
304                )?
305            }
306            _ => {
307                coord_bail!("cannot EXPLAIN {} FOR MATERIALIZED VIEW", stage);
308            }
309        };
310
311        let row = Row::pack_slice(&[Datum::from(explain.as_str())]);
312
313        Ok(Self::send_immediate_rows(row))
314    }
315
316    #[instrument]
317    fn create_materialized_view_validate(
318        &mut self,
319        session: &Session,
320        plan: plan::CreateMaterializedViewPlan,
321        resolved_ids: ResolvedIds,
322        // An optional context set iff the state machine is initiated from
323        // sequencing an EXPLAIN for this statement.
324        explain_ctx: ExplainContext,
325    ) -> Result<CreateMaterializedViewStage, AdapterError> {
326        let plan::CreateMaterializedViewPlan {
327            materialized_view:
328                plan::MaterializedView {
329                    expr,
330                    cluster_id,
331                    refresh_schedule,
332                    ..
333                },
334            ambiguous_columns,
335            ..
336        } = &plan;
337
338        // Validate any references in the materialized view's expression. We do
339        // this on the unoptimized plan to better reflect what the user typed.
340        // We want to reject queries that depend on log sources, for example,
341        // even if we can *technically* optimize that reference away.
342        let expr_depends_on = expr.depends_on();
343        self.catalog()
344            .validate_timeline_context(expr_depends_on.iter().copied())?;
345        self.validate_system_column_references(*ambiguous_columns, &expr_depends_on)?;
346        // Materialized views are not allowed to depend on log sources, as replicas
347        // are not producing the same definite collection for these.
348        let log_names = expr_depends_on
349            .iter()
350            .map(|gid| self.catalog.resolve_item_id(gid))
351            .flat_map(|item_id| self.catalog().introspection_dependencies(item_id))
352            .map(|item_id| self.catalog().get_entry(&item_id).name().item.clone())
353            .collect::<Vec<_>>();
354        if !log_names.is_empty() {
355            return Err(AdapterError::InvalidLogDependency {
356                object_type: "materialized view".into(),
357                log_names,
358            });
359        }
360
361        let validity =
362            PlanValidity::require_transient_revision(self.catalog().transient_revision());
363
364        // Check whether we can read all inputs at all the REFRESH AT times.
365        if let Some(refresh_schedule) = refresh_schedule {
366            if !refresh_schedule.ats.is_empty() && matches!(explain_ctx, ExplainContext::None) {
367                // Purification has acquired the earliest possible read holds if there are any
368                // REFRESH options.
369                let read_holds = self
370                    .txn_read_holds
371                    .get(session.conn_id())
372                    .expect("purification acquired read holds if there are REFRESH ATs");
373                let least_valid_read = read_holds.least_valid_read();
374                for refresh_at_ts in &refresh_schedule.ats {
375                    if !least_valid_read.less_equal(refresh_at_ts) {
376                        return Err(AdapterError::InputNotReadableAtRefreshAtTime(
377                            *refresh_at_ts,
378                            least_valid_read,
379                        ));
380                    }
381                }
382                // Also check that no new id has appeared in `sufficient_collections` (e.g. a new
383                // index), otherwise we might be missing some read holds.
384                let ids = self
385                    .index_oracle(*cluster_id)
386                    .sufficient_collections(resolved_ids.collections().copied());
387                if !ids.difference(&read_holds.id_bundle()).is_empty() {
388                    return Err(AdapterError::ChangedPlan(
389                        "the set of possible inputs changed during the creation of the \
390                         materialized view"
391                            .to_string(),
392                    ));
393                }
394            }
395        }
396
397        Ok(CreateMaterializedViewStage::Optimize(
398            CreateMaterializedViewOptimize {
399                validity,
400                plan,
401                resolved_ids,
402                explain_ctx,
403            },
404        ))
405    }
406
407    #[instrument]
408    async fn create_materialized_view_optimize(
409        &mut self,
410        CreateMaterializedViewOptimize {
411            validity,
412            plan,
413            resolved_ids,
414            explain_ctx,
415        }: CreateMaterializedViewOptimize,
416    ) -> Result<StageResult<Box<CreateMaterializedViewStage>>, AdapterError> {
417        let plan::CreateMaterializedViewPlan {
418            name,
419            materialized_view:
420                plan::MaterializedView {
421                    column_names,
422                    cluster_id,
423                    non_null_assertions,
424                    refresh_schedule,
425                    ..
426                },
427            ..
428        } = &plan;
429
430        // Collect optimizer parameters.
431        let compute_instance = self
432            .instance_snapshot(*cluster_id)
433            .expect("compute instance does not exist");
434        let (item_id, global_id) = if let ExplainContext::None = explain_ctx {
435            let id_ts = self.get_catalog_write_ts().await;
436            self.catalog_mut().allocate_user_id(id_ts).await?
437        } else {
438            self.allocate_transient_id()
439        };
440
441        let (_, view_id) = self.allocate_transient_id();
442        let debug_name = self.catalog().resolve_full_name(name, None).to_string();
443        let optimizer_config = optimize::OptimizerConfig::from(self.catalog().system_config())
444            .override_from(&self.catalog.get_cluster(*cluster_id).config.features())
445            .override_from(&explain_ctx);
446        let force_non_monotonic = Default::default();
447
448        // Build an optimizer for this MATERIALIZED VIEW.
449        let mut optimizer = optimize::materialized_view::Optimizer::new(
450            self.owned_catalog().as_optimizer_catalog(),
451            compute_instance,
452            global_id,
453            view_id,
454            column_names.clone(),
455            non_null_assertions.clone(),
456            refresh_schedule.clone(),
457            debug_name,
458            optimizer_config,
459            self.optimizer_metrics(),
460            force_non_monotonic,
461        );
462
463        let span = Span::current();
464        Ok(StageResult::Handle(mz_ore::task::spawn_blocking(
465            || "optimize create materialized view",
466            move || {
467                span.in_scope(|| {
468                    let mut pipeline = || -> Result<(
469                        optimize::materialized_view::LocalMirPlan,
470                        optimize::materialized_view::GlobalMirPlan,
471                        optimize::materialized_view::GlobalLirPlan,
472                    ), AdapterError> {
473                        let _dispatch_guard = explain_ctx.dispatch_guard();
474
475                        let raw_expr = plan.materialized_view.expr.clone();
476
477                        // HIR ⇒ MIR lowering and MIR ⇒ MIR optimization (local and global)
478                        let local_mir_plan = optimizer.catch_unwind_optimize(raw_expr)?;
479                        let global_mir_plan = optimizer.catch_unwind_optimize(local_mir_plan.clone())?;
480                        // MIR ⇒ LIR lowering and LIR ⇒ LIR optimization (global)
481                        let global_lir_plan = optimizer.catch_unwind_optimize(global_mir_plan.clone())?;
482
483                        Ok((local_mir_plan, global_mir_plan, global_lir_plan))
484                    };
485
486                    let stage = match pipeline() {
487                        Ok((local_mir_plan, global_mir_plan, global_lir_plan)) => {
488                            if let ExplainContext::Plan(explain_ctx) = explain_ctx {
489                                let (_, df_meta) = global_lir_plan.unapply();
490                                CreateMaterializedViewStage::Explain(
491                                    CreateMaterializedViewExplain {
492                                        validity,
493                                        global_id,
494                                        plan,
495                                        df_meta,
496                                        explain_ctx,
497                                    },
498                                )
499                            } else {
500                                CreateMaterializedViewStage::Finish(CreateMaterializedViewFinish {
501                                    item_id,
502                                    global_id,
503                                    validity,
504                                    plan,
505                                    resolved_ids,
506                                    local_mir_plan,
507                                    global_mir_plan,
508                                    global_lir_plan,
509                                })
510                            }
511                        }
512                        // Internal optimizer errors are handled differently
513                        // depending on the caller.
514                        Err(err) => {
515                            let ExplainContext::Plan(explain_ctx) = explain_ctx else {
516                                // In `sequence_~` contexts, immediately return the error.
517                                return Err(err);
518                            };
519
520                            if explain_ctx.broken {
521                                // In `EXPLAIN BROKEN` contexts, just log the error
522                                // and move to the next stage with default
523                                // parameters.
524                                tracing::error!("error while handling EXPLAIN statement: {}", err);
525                                CreateMaterializedViewStage::Explain(
526                                    CreateMaterializedViewExplain {
527                                        global_id,
528                                        validity,
529                                        plan,
530                                        df_meta: Default::default(),
531                                        explain_ctx,
532                                    },
533                                )
534                            } else {
535                                // In regular `EXPLAIN` contexts, immediately return the error.
536                                return Err(err);
537                            }
538                        }
539                    };
540
541                    Ok(Box::new(stage))
542                })
543            },
544        )))
545    }
546
547    #[instrument]
548    async fn create_materialized_view_finish(
549        &mut self,
550        ctx: &mut ExecuteContext,
551        stage: CreateMaterializedViewFinish,
552    ) -> Result<StageResult<Box<CreateMaterializedViewStage>>, AdapterError> {
553        let CreateMaterializedViewFinish {
554            item_id,
555            global_id,
556            plan:
557                plan::CreateMaterializedViewPlan {
558                    name,
559                    materialized_view:
560                        plan::MaterializedView {
561                            mut create_sql,
562                            expr: raw_expr,
563                            dependencies,
564                            cluster_id,
565                            non_null_assertions,
566                            compaction_window,
567                            refresh_schedule,
568                            ..
569                        },
570                    drop_ids,
571                    if_not_exists,
572                    ..
573                },
574            resolved_ids,
575            local_mir_plan,
576            global_mir_plan,
577            global_lir_plan,
578            ..
579        } = stage;
580        // Timestamp selection
581        let id_bundle = dataflow_import_id_bundle(global_lir_plan.df_desc(), cluster_id);
582
583        let read_holds_owned;
584        let read_holds = if let Some(txn_reads) = self.txn_read_holds.get(ctx.session().conn_id()) {
585            // In some cases, for example when REFRESH is used, the preparatory
586            // stages will already have acquired ReadHolds, we can re-use those.
587
588            txn_reads
589        } else {
590            // No one has acquired holds, make sure we can determine an as_of
591            // and render our dataflow below.
592            read_holds_owned = self.acquire_read_holds(&id_bundle);
593            &read_holds_owned
594        };
595
596        let (dataflow_as_of, storage_as_of, until) =
597            self.select_timestamps(id_bundle, refresh_schedule.as_ref(), read_holds)?;
598
599        tracing::info!(
600            dataflow_as_of = ?dataflow_as_of,
601            storage_as_of = ?storage_as_of,
602            until = ?until,
603            "materialized view timestamp selection",
604        );
605
606        let initial_as_of = storage_as_of.clone();
607
608        // Update the `create_sql` with the selected `as_of`. This is how we make sure the `as_of`
609        // is persisted to the catalog and can be relied on during bootstrapping.
610        // This has to be the `storage_as_of`, because bootstrapping uses this in
611        // `bootstrap_storage_collections`.
612        if let Some(storage_as_of_ts) = storage_as_of.as_option() {
613            let stmt = mz_sql::parse::parse(&create_sql)
614                .map_err(|_| {
615                    AdapterError::internal(
616                        "create materialized view",
617                        "original SQL should roundtrip",
618                    )
619                })?
620                .into_element()
621                .ast;
622            let ast::Statement::CreateMaterializedView(mut stmt) = stmt else {
623                panic!("unexpected statement type");
624            };
625            stmt.as_of = Some(storage_as_of_ts.into());
626            create_sql = stmt.to_ast_string_stable();
627        }
628
629        let ops = vec![
630            catalog::Op::DropObjects(
631                drop_ids
632                    .into_iter()
633                    .map(catalog::DropObjectInfo::Item)
634                    .collect(),
635            ),
636            catalog::Op::CreateItem {
637                id: item_id,
638                name: name.clone(),
639                item: CatalogItem::MaterializedView(MaterializedView {
640                    create_sql,
641                    raw_expr: raw_expr.into(),
642                    optimized_expr: local_mir_plan.expr().into(),
643                    desc: global_lir_plan.desc().clone(),
644                    global_id,
645                    resolved_ids,
646                    dependencies,
647                    cluster_id,
648                    non_null_assertions,
649                    custom_logical_compaction_window: compaction_window,
650                    refresh_schedule: refresh_schedule.clone(),
651                    initial_as_of: Some(initial_as_of.clone()),
652                }),
653                owner_id: *ctx.session().current_role_id(),
654            },
655        ];
656
657        // Pre-allocate a vector of transient GlobalIds for each notice.
658        let notice_ids = std::iter::repeat_with(|| self.allocate_transient_id())
659            .map(|(_item_id, global_id)| global_id)
660            .take(global_lir_plan.df_meta().optimizer_notices.len())
661            .collect::<Vec<_>>();
662
663        let transact_result = self
664            .catalog_transact_with_side_effects(Some(ctx), ops, move |coord, ctx| {
665                Box::pin(async move {
666                    let output_desc = global_lir_plan.desc().clone();
667                    let (mut df_desc, df_meta) = global_lir_plan.unapply();
668
669                    // Save plan structures.
670                    coord
671                        .catalog_mut()
672                        .set_optimized_plan(global_id, global_mir_plan.df_desc().clone());
673                    coord
674                        .catalog_mut()
675                        .set_physical_plan(global_id, df_desc.clone());
676
677                    let notice_builtin_updates_fut = coord
678                        .process_dataflow_metainfo(df_meta, global_id, ctx, notice_ids)
679                        .await;
680
681                    df_desc.set_as_of(dataflow_as_of.clone());
682                    df_desc.set_initial_as_of(initial_as_of);
683                    df_desc.until = until;
684
685                    let storage_metadata = coord.catalog.state().storage_metadata();
686
687                    // Announce the creation of the materialized view source.
688                    coord
689                        .controller
690                        .storage
691                        .create_collections(
692                            storage_metadata,
693                            None,
694                            vec![(
695                                global_id,
696                                CollectionDescription {
697                                    desc: output_desc,
698                                    data_source: DataSource::Other,
699                                    since: Some(storage_as_of),
700                                    status_collection_id: None,
701                                    timeline: None,
702                                },
703                            )],
704                        )
705                        .await
706                        .unwrap_or_terminate("cannot fail to append");
707
708                    coord
709                        .initialize_storage_read_policies(
710                            btreeset![item_id],
711                            compaction_window.unwrap_or(CompactionWindow::Default),
712                        )
713                        .await;
714
715                    coord
716                        .ship_dataflow_and_notice_builtin_table_updates(
717                            df_desc,
718                            cluster_id,
719                            notice_builtin_updates_fut,
720                        )
721                        .await;
722                })
723            })
724            .await;
725
726        match transact_result {
727            Ok(_) => Ok(ExecuteResponse::CreatedMaterializedView),
728            Err(AdapterError::Catalog(mz_catalog::memory::error::Error {
729                kind:
730                    mz_catalog::memory::error::ErrorKind::Sql(CatalogError::ItemAlreadyExists(_, _)),
731            })) if if_not_exists => {
732                ctx.session()
733                    .add_notice(AdapterNotice::ObjectAlreadyExists {
734                        name: name.item,
735                        ty: "materialized view",
736                    });
737                Ok(ExecuteResponse::CreatedMaterializedView)
738            }
739            Err(err) => Err(err),
740        }
741        .map(StageResult::Response)
742    }
743
744    /// Select the initial `dataflow_as_of`, `storage_as_of`, and `until` frontiers for a
745    /// materialized view.
746    fn select_timestamps(
747        &self,
748        id_bundle: CollectionIdBundle,
749        refresh_schedule: Option<&RefreshSchedule>,
750        read_holds: &ReadHolds<mz_repr::Timestamp>,
751    ) -> Result<
752        (
753            Antichain<mz_repr::Timestamp>,
754            Antichain<mz_repr::Timestamp>,
755            Antichain<mz_repr::Timestamp>,
756        ),
757        AdapterError,
758    > {
759        assert!(
760            id_bundle.difference(&read_holds.id_bundle()).is_empty(),
761            "we must have read holds for all involved collections"
762        );
763
764        // For non-REFRESH MVs both the `dataflow_as_of` and the `storage_as_of` should be simply
765        // `least_valid_read`.
766        let least_valid_read = read_holds.least_valid_read();
767        let mut dataflow_as_of = least_valid_read.clone();
768        let mut storage_as_of = least_valid_read.clone();
769
770        // For MVs with non-trivial REFRESH schedules:
771        // 1. it's important to set the `storage_as_of` to the first refresh. This is because we'd
772        // like queries on the MV to block until the first refresh (rather than to show an empty
773        // MV).
774        // 2. We move the `dataflow_as_of` forward to the minimum of `greatest_available_read` and
775        // the first refresh time. There is no point in processing the times before
776        // `greatest_available_read`, because the first time for which results will be exposed is
777        // the first refresh time. Also note that simply moving the `dataflow_as_of` forward to the
778        // first refresh time would prevent warmup before the first refresh.
779        if let Some(refresh_schedule) = &refresh_schedule {
780            if let Some(least_valid_read_ts) = least_valid_read.as_option() {
781                if let Some(first_refresh_ts) =
782                    refresh_schedule.round_up_timestamp(*least_valid_read_ts)
783                {
784                    storage_as_of = Antichain::from_elem(first_refresh_ts);
785                    dataflow_as_of.join_assign(
786                        &self
787                            .greatest_available_read(&id_bundle)
788                            .meet(&storage_as_of),
789                    );
790                } else {
791                    let last_refresh = refresh_schedule.last_refresh().expect(
792                        "if round_up_timestamp returned None, then there should be a last refresh",
793                    );
794
795                    return Err(AdapterError::MaterializedViewWouldNeverRefresh(
796                        last_refresh,
797                        *least_valid_read_ts,
798                    ));
799                }
800            } else {
801                // The `as_of` should never be empty, because then the MV would be unreadable.
802                soft_panic_or_log!("creating a materialized view with an empty `as_of`");
803            }
804        }
805
806        // If we have a refresh schedule that has a last refresh, then set the `until` to the last refresh.
807        // (If the `try_step_forward` fails, then no need to set an `until`, because it's not possible to get any data
808        // beyond that last refresh time, because there are no times beyond that time.)
809        let until_ts = refresh_schedule
810            .and_then(|s| s.last_refresh())
811            .and_then(|r| r.try_step_forward());
812        let until = Antichain::from_iter(until_ts);
813
814        Ok((dataflow_as_of, storage_as_of, until))
815    }
816
817    #[instrument]
818    async fn create_materialized_view_explain(
819        &mut self,
820        session: &Session,
821        CreateMaterializedViewExplain {
822            global_id,
823            plan:
824                plan::CreateMaterializedViewPlan {
825                    name,
826                    materialized_view:
827                        plan::MaterializedView {
828                            column_names,
829                            cluster_id,
830                            ..
831                        },
832                    ..
833                },
834            df_meta,
835            explain_ctx:
836                ExplainPlanContext {
837                    config,
838                    format,
839                    stage,
840                    optimizer_trace,
841                    ..
842                },
843            ..
844        }: CreateMaterializedViewExplain,
845    ) -> Result<StageResult<Box<CreateMaterializedViewStage>>, AdapterError> {
846        let session_catalog = self.catalog().for_session(session);
847        let expr_humanizer = {
848            let full_name = self.catalog().resolve_full_name(&name, None);
849            let transient_items = btreemap! {
850                global_id => TransientItem::new(
851                    Some(full_name.into_parts()),
852                    Some(column_names.iter().map(|c| c.to_string()).collect()),
853                )
854            };
855            ExprHumanizerExt::new(transient_items, &session_catalog)
856        };
857
858        let target_cluster = self.catalog().get_cluster(cluster_id);
859
860        let features = OptimizerFeatures::from(self.catalog().system_config())
861            .override_from(&target_cluster.config.features())
862            .override_from(&config.features);
863
864        let rows = optimizer_trace
865            .into_rows(
866                format,
867                &config,
868                &features,
869                &expr_humanizer,
870                None,
871                Some(target_cluster),
872                df_meta,
873                stage,
874                plan::ExplaineeStatementKind::CreateMaterializedView,
875                None,
876            )
877            .await?;
878
879        Ok(StageResult::Response(Self::send_immediate_rows(rows)))
880    }
881
882    pub(crate) async fn explain_pushdown_materialized_view(
883        &self,
884        ctx: ExecuteContext,
885        item_id: CatalogItemId,
886    ) {
887        let CatalogItem::MaterializedView(mview) = self.catalog().get_entry(&item_id).item() else {
888            unreachable!() // Asserted in `sequence_explain_pushdown`.
889        };
890        let gid = mview.global_id();
891        let mview = mview.clone();
892
893        let Some(plan) = self.catalog().try_get_physical_plan(&gid).cloned() else {
894            let msg = format!("cannot find plan for materialized view {item_id} in catalog");
895            tracing::error!("{msg}");
896            ctx.retire(Err(anyhow!("{msg}").into()));
897            return;
898        };
899
900        // We don't have any way to "duplicate" the read hold of the actual collection, which we
901        // obtain below... but the current implementation of read holds guarantees that the storage
902        // holds we obtain here will not be any greater than the hold we actually want.
903        let read_holds =
904            Some(self.acquire_read_holds(&dataflow_import_id_bundle(&plan, mview.cluster_id)));
905
906        let frontiers = self
907            .controller
908            .compute
909            .collection_frontiers(gid, Some(mview.cluster_id))
910            .expect("materialized view exists");
911
912        let as_of = frontiers.read_frontier.to_owned();
913
914        let until = mview
915            .refresh_schedule
916            .as_ref()
917            .and_then(|s| s.last_refresh())
918            .unwrap_or(mz_repr::Timestamp::MAX);
919
920        let mz_now = match as_of.as_option() {
921            Some(&as_of) => {
922                ResultSpec::value_between(Datum::MzTimestamp(as_of), Datum::MzTimestamp(until))
923            }
924            None => ResultSpec::value_all(),
925        };
926
927        self.render_explain_pushdown(
928            ctx,
929            as_of,
930            mz_now,
931            read_holds,
932            plan.source_imports
933                .into_iter()
934                .filter_map(|(id, (source, _, _upper))| {
935                    source.arguments.operators.map(|mfp| (id, mfp))
936                }),
937        )
938        .await
939    }
940}