mz_adapter/optimize/
pinned.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//! A pinned LIR plan is a durable record of a dataflow's plan;
11//! we must _instantiate_ that plan with other runtime information
12//! to actually run the dataflow.
13//!
14//! [`CatalogInstantiationContext`] is the [`InstantiationContext`] for
15//! `PinnedDataflow::instantiate`, supplying the export descriptors not
16//! recorded in the pinned plan and the current shape of each import. We look
17//! this information up in the catalog items; each lookup reproduces what we
18//! would have built in the first place.
19
20use mz_catalog::memory::objects::CatalogItem;
21use mz_compute_types::plan::pinned::{
22    IndexInfo, InstantiationContext, MaterializedViewInfo, MetricSinkInfo,
23};
24use mz_expr::MirRelationExpr;
25use mz_repr::global_id::TransientIdGen;
26use mz_repr::{GlobalId, RelationVersionSelector, ReprRelationType, SqlRelationType};
27
28use crate::catalog::CatalogState;
29use crate::optimize::metric_sink::shape_metric_sink_source;
30
31/// The catalog and the coordinator's transient id generator, which together
32/// answer everything instantiation asks for.
33// TODO(mgree): construct this where the coordinator loads pinned plans.
34#[cfg_attr(not(test), allow(dead_code))]
35#[derive(Debug, Clone, Copy)]
36pub struct CatalogInstantiationContext<'a> {
37    pub catalog: &'a CatalogState,
38    pub transient_ids: &'a TransientIdGen,
39}
40
41impl InstantiationContext for CatalogInstantiationContext<'_> {
42    fn collection_type(&self, id: GlobalId) -> Option<SqlRelationType> {
43        let entry = self.catalog.try_get_entry_by_global_id(&id)?;
44        // Tables and materialized views are imported at the version `id`
45        // names, as `DataflowBuilder::import_into_dataflow` does.
46        let typ = match entry.item() {
47            CatalogItem::Table(table) => table.desc_for(&id).into_typ(),
48            CatalogItem::MaterializedView(mv) => mv.desc_for(&id).into_typ(),
49            other => other
50                .relation_desc(RelationVersionSelector::Latest)?
51                .typ()
52                .clone(),
53        };
54        Some(typ)
55    }
56
57    fn index(&self, id: GlobalId) -> Option<IndexInfo> {
58        let index = self.catalog.try_get_entry_by_global_id(&id)?.index()?;
59        let on_desc = self.catalog.try_get_desc_by_global_id(&index.on)?;
60        Some(IndexInfo {
61            on: index.on,
62            keys: index.keys.to_vec(),
63            typ: ReprRelationType::from(on_desc.typ()),
64        })
65    }
66
67    fn materialized_view(&self, id: GlobalId) -> Option<MaterializedViewInfo> {
68        let mv = self
69            .catalog
70            .try_get_entry_by_global_id(&id)?
71            .materialized_view()?;
72        // Only the writing collection is fed by a sink. An older version's id
73        // resolves to the same item but names no sink.
74        if mv.global_id_writes() != id {
75            return None;
76        }
77        Some(MaterializedViewInfo {
78            desc: mv.desc.latest(),
79            non_null_assertions: mv.non_null_assertions.clone(),
80            refresh_schedule: mv.refresh_schedule.clone(),
81        })
82    }
83
84    fn metric_sink(&self, id: GlobalId) -> Option<MetricSinkInfo> {
85        let CatalogItem::MetricSink(sink) = self.catalog.try_get_entry_by_global_id(&id)?.item()
86        else {
87            return None;
88        };
89        let from_desc = self.catalog.try_get_desc_by_global_id(&sink.from)?;
90        // The sink reads the shaped view the optimizer builds over `from`,
91        // whose description depends only on the source description and the
92        // prefix. The expression the shaping also returns is not needed here.
93        let source =
94            MirRelationExpr::global_get(sink.from, ReprRelationType::from(from_desc.typ()));
95        let (_shaped_expr, shaped_desc) =
96            shape_metric_sink_source(source, &from_desc, &sink.prefix);
97        Some(MetricSinkInfo {
98            // A user sink passes no label to the optimizer, which then labels
99            // the sink with its id.
100            label: id.to_string(),
101            from_desc: shaped_desc,
102        })
103    }
104
105    fn allocate_transient_id(&self) -> GlobalId {
106        self.transient_ids.allocate_id().1
107    }
108}
109
110#[cfg(test)]
111#[path = "pinned_tests.rs"]
112mod tests;
mz_adapter/optimize/pinned.rs

mz_adapter/optimize/
pinned.rs
