1use 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 unreachable!()
134 };
135 let plan::ExplaineeStatement::CreateMaterializedView { broken, plan } = stmt else {
136 unreachable!()
139 };
140
141 let optimizer_trace = OptimizerTrace::new(stage.paths());
144
145 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!() };
178 let CatalogItem::MaterializedView(item) = self.catalog().get_entry(&id).item() else {
179 unreachable!() };
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!() };
192
193 let broken = false;
196
197 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!() };
231 let CatalogItem::MaterializedView(view) = self.catalog().get_entry(&id).item() else {
232 unreachable!() };
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 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 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 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 if let Some(refresh_schedule) = refresh_schedule {
366 if !refresh_schedule.ats.is_empty() && matches!(explain_ctx, ExplainContext::None) {
367 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 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 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 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 let local_mir_plan = optimizer.catch_unwind_optimize(raw_expr)?;
479 let global_mir_plan = optimizer.catch_unwind_optimize(local_mir_plan.clone())?;
480 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 Err(err) => {
515 let ExplainContext::Plan(explain_ctx) = explain_ctx else {
516 return Err(err);
518 };
519
520 if explain_ctx.broken {
521 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 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 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 txn_reads
589 } else {
590 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 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 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 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 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 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 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 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 soft_panic_or_log!("creating a materialized view with an empty `as_of`");
803 }
804 }
805
806 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!() };
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 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}