mz_adapter/optimize/
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
10//! Optimizer implementation for `CREATE MATERIALIZED VIEW` statements.
11//!
12//! Note that, in contrast to other optimization pipelines, timestamp selection is not part of
13//! MV optimization. Instead users are expected to separately set the as-of on the optimized
14//! `DataflowDescription` received from `GlobalLirPlan::unapply`. Reasons for choosing to exclude
15//! timestamp selection from the MV optimization pipeline are:
16//!
17//!  (a) MVs don't support non-empty `until` frontiers, so they don't provide opportunity for
18//!      optimizations based on the selected timestamp.
19//!  (b) We want to generate dataflow plans early during environment bootstrapping, before we have
20//!      access to all information required for timestamp selection.
21//!
22//! None of this is set in stone though. If we find an opportunity for optimizing MVs based on
23//! their timestamps, we'll want to make timestamp selection part of the MV optimization again and
24//! find a different approach to bootstrapping.
25//!
26//! See also MaterializeInc/materialize#22940.
27
28use std::collections::BTreeSet;
29use std::sync::Arc;
30use std::time::{Duration, Instant};
31
32use mz_compute_types::plan::Plan;
33use mz_compute_types::sinks::{
34    ComputeSinkConnection, ComputeSinkDesc, MaterializedViewSinkConnection,
35};
36use mz_expr::{MirRelationExpr, OptimizedMirRelationExpr};
37use mz_repr::explain::trace_plan;
38use mz_repr::refresh_schedule::RefreshSchedule;
39use mz_repr::{ColumnName, GlobalId, RelationDesc};
40use mz_sql::optimizer_metrics::OptimizerMetrics;
41use mz_sql::plan::HirRelationExpr;
42use mz_transform::TransformCtx;
43use mz_transform::dataflow::DataflowMetainfo;
44use mz_transform::normalize_lets::normalize_lets;
45use mz_transform::typecheck::{SharedContext as TypecheckContext, empty_context};
46use timely::progress::Antichain;
47
48use crate::optimize::dataflows::{
49    ComputeInstanceSnapshot, DataflowBuilder, ExprPrepStyle, prep_relation_expr, prep_scalar_expr,
50};
51use crate::optimize::{
52    LirDataflowDescription, MirDataflowDescription, Optimize, OptimizeMode, OptimizerCatalog,
53    OptimizerConfig, OptimizerError, optimize_mir_local, trace_plan,
54};
55
56pub struct Optimizer {
57    /// A typechecking context to use throughout the optimizer pipeline.
58    typecheck_ctx: TypecheckContext,
59    /// A snapshot of the catalog state.
60    catalog: Arc<dyn OptimizerCatalog>,
61    /// A snapshot of the cluster that will run the dataflows.
62    compute_instance: ComputeInstanceSnapshot,
63    /// A durable GlobalId to be used with the exported materialized view sink.
64    sink_id: GlobalId,
65    /// A transient GlobalId to be used when constructing the dataflow.
66    view_id: GlobalId,
67    /// The resulting column names.
68    column_names: Vec<ColumnName>,
69    /// Output columns that are asserted to be not null in the `CREATE VIEW`
70    /// statement.
71    non_null_assertions: Vec<usize>,
72    /// Refresh schedule, e.g., `REFRESH EVERY '1 day'`
73    refresh_schedule: Option<RefreshSchedule>,
74    /// A human-readable name exposed internally (useful for debugging).
75    debug_name: String,
76    /// Optimizer config.
77    config: OptimizerConfig,
78    /// Optimizer metrics.
79    metrics: OptimizerMetrics,
80    /// The time spent performing optimization so far.
81    duration: Duration,
82    /// Overrides monotonicity for the given source collections.
83    ///
84    /// This is here only for continual tasks, which at runtime introduce
85    /// synthetic retractions to "input sources". If/when we split a CT
86    /// optimizer out of the MV optimizer, this can be removed.
87    ///
88    /// TODO(ct3): There are other differences between a GlobalId used as a CT
89    /// input vs as a normal collection, such as the statistical size estimates.
90    /// Plus, at the moment, it is not possible to use the same GlobalId as both
91    /// an "input" and a "reference" in a CT. So, better than this approach
92    /// would be for the optimizer itself to somehow understand the distinction
93    /// between a CT input and a normal collection.
94    ///
95    /// In the meantime, it might be desirable to refactor the MV optimizer to
96    /// have a small amount of knowledge about CTs, in particular producing the
97    /// CT sink connection directly. This would allow us to replace this field
98    /// with something derived directly from that sink connection.
99    force_source_non_monotonic: BTreeSet<GlobalId>,
100}
101
102impl Optimizer {
103    pub fn new(
104        catalog: Arc<dyn OptimizerCatalog>,
105        compute_instance: ComputeInstanceSnapshot,
106        sink_id: GlobalId,
107        view_id: GlobalId,
108        column_names: Vec<ColumnName>,
109        non_null_assertions: Vec<usize>,
110        refresh_schedule: Option<RefreshSchedule>,
111        debug_name: String,
112        config: OptimizerConfig,
113        metrics: OptimizerMetrics,
114        force_source_non_monotonic: BTreeSet<GlobalId>,
115    ) -> Self {
116        Self {
117            typecheck_ctx: empty_context(),
118            catalog,
119            compute_instance,
120            sink_id,
121            view_id,
122            column_names,
123            non_null_assertions,
124            refresh_schedule,
125            debug_name,
126            config,
127            metrics,
128            duration: Default::default(),
129            force_source_non_monotonic,
130        }
131    }
132}
133
134/// The (sealed intermediate) result after HIR ⇒ MIR lowering and decorrelation
135/// and MIR optimization.
136#[derive(Clone, Debug)]
137pub struct LocalMirPlan {
138    expr: MirRelationExpr,
139    df_meta: DataflowMetainfo,
140}
141
142/// The (sealed intermediate) result after:
143///
144/// 1. embedding a [`LocalMirPlan`] into a [`MirDataflowDescription`],
145/// 2. transitively inlining referenced views, and
146/// 3. jointly optimizing the `MIR` plans in the [`MirDataflowDescription`].
147#[derive(Clone, Debug)]
148pub struct GlobalMirPlan {
149    df_desc: MirDataflowDescription,
150    df_meta: DataflowMetainfo,
151}
152
153impl GlobalMirPlan {
154    pub fn df_desc(&self) -> &MirDataflowDescription {
155        &self.df_desc
156    }
157}
158
159/// The (final) result after MIR ⇒ LIR lowering and optimizing the resulting
160/// `DataflowDescription` with `LIR` plans.
161#[derive(Clone, Debug)]
162pub struct GlobalLirPlan {
163    df_desc: LirDataflowDescription,
164    df_meta: DataflowMetainfo,
165}
166
167impl GlobalLirPlan {
168    pub fn df_desc(&self) -> &LirDataflowDescription {
169        &self.df_desc
170    }
171
172    pub fn df_meta(&self) -> &DataflowMetainfo {
173        &self.df_meta
174    }
175
176    pub fn desc(&self) -> &RelationDesc {
177        let sink_exports = &self.df_desc.sink_exports;
178        let sink = sink_exports.values().next().expect("valid sink");
179        &sink.from_desc
180    }
181}
182
183impl Optimize<HirRelationExpr> for Optimizer {
184    type To = LocalMirPlan;
185
186    fn optimize(&mut self, expr: HirRelationExpr) -> Result<Self::To, OptimizerError> {
187        let time = Instant::now();
188
189        // Trace the pipeline input under `optimize/raw`.
190        trace_plan!(at: "raw", &expr);
191
192        // HIR ⇒ MIR lowering and decorrelation
193        let expr = expr.lower(&self.config, Some(&self.metrics))?;
194
195        // MIR ⇒ MIR optimization (local)
196        let mut df_meta = DataflowMetainfo::default();
197        let mut transform_ctx = TransformCtx::local(
198            &self.config.features,
199            &self.typecheck_ctx,
200            &mut df_meta,
201            Some(&self.metrics),
202            Some(self.view_id),
203        );
204        let expr = optimize_mir_local(expr, &mut transform_ctx)?.into_inner();
205
206        self.duration += time.elapsed();
207
208        // Return the (sealed) plan at the end of this optimization step.
209        Ok(LocalMirPlan { expr, df_meta })
210    }
211}
212
213impl LocalMirPlan {
214    pub fn expr(&self) -> OptimizedMirRelationExpr {
215        OptimizedMirRelationExpr(self.expr.clone())
216    }
217}
218
219/// This is needed only because the pipeline in the bootstrap code starts from an
220/// [`OptimizedMirRelationExpr`] attached to a [`mz_catalog::memory::objects::CatalogItem`].
221impl Optimize<OptimizedMirRelationExpr> for Optimizer {
222    type To = GlobalMirPlan;
223
224    fn optimize(&mut self, expr: OptimizedMirRelationExpr) -> Result<Self::To, OptimizerError> {
225        let expr = expr.into_inner();
226        let df_meta = DataflowMetainfo::default();
227        self.optimize(LocalMirPlan { expr, df_meta })
228    }
229}
230
231impl Optimize<LocalMirPlan> for Optimizer {
232    type To = GlobalMirPlan;
233
234    fn optimize(&mut self, plan: LocalMirPlan) -> Result<Self::To, OptimizerError> {
235        let time = Instant::now();
236
237        let expr = OptimizedMirRelationExpr(plan.expr);
238        let mut df_meta = plan.df_meta;
239
240        let mut rel_typ = expr.typ();
241        for &i in self.non_null_assertions.iter() {
242            rel_typ.column_types[i].nullable = false;
243        }
244        let rel_desc = RelationDesc::new(rel_typ, self.column_names.clone());
245
246        let mut df_builder = {
247            let compute = self.compute_instance.clone();
248            DataflowBuilder::new(&*self.catalog, compute).with_config(&self.config)
249        };
250        let mut df_desc = MirDataflowDescription::new(self.debug_name.clone());
251
252        df_desc.refresh_schedule.clone_from(&self.refresh_schedule);
253
254        df_builder.import_view_into_dataflow(
255            &self.view_id,
256            &expr,
257            &mut df_desc,
258            &self.config.features,
259        )?;
260        df_builder.maybe_reoptimize_imported_views(&mut df_desc, &self.config)?;
261
262        let sink_description = ComputeSinkDesc {
263            from: self.view_id,
264            from_desc: rel_desc.clone(),
265            connection: ComputeSinkConnection::MaterializedView(MaterializedViewSinkConnection {
266                value_desc: rel_desc,
267                storage_metadata: (),
268            }),
269            with_snapshot: true,
270            up_to: Antichain::default(),
271            non_null_assertions: self.non_null_assertions.clone(),
272            refresh_schedule: self.refresh_schedule.clone(),
273        };
274        df_desc.export_sink(self.sink_id, sink_description);
275
276        // Prepare expressions in the assembled dataflow.
277        let style = ExprPrepStyle::Maintained;
278        df_desc.visit_children(
279            |r| prep_relation_expr(r, style),
280            |s| prep_scalar_expr(s, style),
281        )?;
282
283        // Construct TransformCtx for global optimization.
284        let mut transform_ctx = TransformCtx::global(
285            &df_builder,
286            &mz_transform::EmptyStatisticsOracle, // TODO: wire proper stats
287            &self.config.features,
288            &self.typecheck_ctx,
289            &mut df_meta,
290            Some(&self.metrics),
291        );
292        // Apply source monotonicity overrides.
293        for id in self.force_source_non_monotonic.iter() {
294            if let Some((_desc, monotonic, _upper)) = df_desc.source_imports.get_mut(id) {
295                *monotonic = false;
296            }
297        }
298        // Run global optimization.
299        mz_transform::optimize_dataflow(&mut df_desc, &mut transform_ctx, false)?;
300
301        if self.config.mode == OptimizeMode::Explain {
302            // Collect the list of indexes used by the dataflow at this point.
303            trace_plan!(at: "global", &df_meta.used_indexes(&df_desc));
304        }
305
306        self.duration += time.elapsed();
307
308        // Return the (sealed) plan at the end of this optimization step.
309        Ok(GlobalMirPlan { df_desc, df_meta })
310    }
311}
312
313impl Optimize<GlobalMirPlan> for Optimizer {
314    type To = GlobalLirPlan;
315
316    fn optimize(&mut self, plan: GlobalMirPlan) -> Result<Self::To, OptimizerError> {
317        let time = Instant::now();
318
319        let GlobalMirPlan {
320            mut df_desc,
321            df_meta,
322        } = plan;
323
324        // Ensure all expressions are normalized before finalizing.
325        for build in df_desc.objects_to_build.iter_mut() {
326            normalize_lets(&mut build.plan.0, &self.config.features)?
327        }
328
329        // Finalize the dataflow. This includes:
330        // - MIR ⇒ LIR lowering
331        // - LIR ⇒ LIR transforms
332        let df_desc = Plan::finalize_dataflow(df_desc, &self.config.features)?;
333
334        // Trace the pipeline output under `optimize`.
335        trace_plan(&df_desc);
336
337        self.duration += time.elapsed();
338        self.metrics
339            .observe_e2e_optimization_time("materialized_view", self.duration);
340
341        // Return the plan at the end of this `optimize` step.
342        Ok(GlobalLirPlan { df_desc, df_meta })
343    }
344}
345
346impl GlobalLirPlan {
347    /// Unwraps the parts of the final result of the optimization pipeline.
348    pub fn unapply(self) -> (LirDataflowDescription, DataflowMetainfo) {
349        (self.df_desc, self.df_meta)
350    }
351}
mz_adapter/optimize/materialized_view.rs

mz_adapter/optimize/
materialized_view.rs
