mz_compute_types/
dataflows.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//! Types for describing dataflows.
11
12use std::collections::{BTreeMap, BTreeSet};
13use std::fmt;
14
15use mz_expr::{CollectionPlan, MirRelationExpr, MirScalarExpr, OptimizedMirRelationExpr};
16use mz_ore::collections::CollectionExt;
17use mz_ore::soft_assert_or_log;
18use mz_repr::refresh_schedule::RefreshSchedule;
19use mz_repr::{GlobalId, ReprRelationType, SqlRelationType, Timestamp};
20use mz_storage_types::time_dependence::TimeDependence;
21use serde::{Deserialize, Serialize};
22use timely::progress::Antichain;
23
24use crate::plan::LirRelationExpr;
25use crate::plan::render_plan::RenderPlan;
26use crate::plan::scalar::{LirScalarExpr, lses_from_mses};
27use crate::sinks::{ComputeSinkConnection, ComputeSinkDesc};
28use crate::sources::{SourceInstanceArguments, SourceInstanceDesc};
29
30/// A description of a dataflow to construct and results to surface.
31#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
32pub struct DataflowDescription<P, S: 'static = ()> {
33    /// Sources instantiations made available to the dataflow pair with monotonicity information.
34    pub source_imports: BTreeMap<GlobalId, SourceImport<S>>,
35    /// Indexes made available to the dataflow.
36    /// (id of index, import)
37    pub index_imports: BTreeMap<GlobalId, IndexImport>,
38    /// Views and indexes to be built and stored in the local context.
39    /// Objects must be built in the specific order, as there may be
40    /// dependencies of later objects on prior identifiers.
41    pub objects_to_build: Vec<BuildDesc<P>>,
42    /// Indexes to be made available to be shared with other dataflows
43    /// (id of new index, description of index, relationtype of base source/view/table)
44    pub index_exports: BTreeMap<GlobalId, (IndexDesc<MirScalarExpr>, ReprRelationType)>,
45    /// sinks to be created
46    /// (id of new sink, description of sink)
47    pub sink_exports: BTreeMap<GlobalId, ComputeSinkDesc<S>>,
48    /// An optional frontier to which inputs should be advanced.
49    ///
50    /// If this is set, it should override the default setting determined by
51    /// the upper bound of `since` frontiers contributing to the dataflow.
52    /// It is an error for this to be set to a frontier not beyond that default.
53    pub as_of: Option<Antichain<Timestamp>>,
54    /// Frontier beyond which the dataflow should not execute.
55    /// Specifically, updates at times greater or equal to this frontier are suppressed.
56    /// This is often set to `as_of + 1` to enable "batch" computations.
57    /// Note that frontier advancements might still happen to times that are after the `until`,
58    /// only data is suppressed. (This is consistent with how frontier advancements can also
59    /// happen before the `as_of`.)
60    pub until: Antichain<Timestamp>,
61    /// The initial as_of when the collection is first created. Filled only for materialized views.
62    /// Note that this doesn't change upon restarts.
63    pub initial_storage_as_of: Option<Antichain<Timestamp>>,
64    /// The schedule of REFRESH materialized views.
65    pub refresh_schedule: Option<RefreshSchedule>,
66    /// Human-readable name
67    pub debug_name: String,
68    /// Description of how the dataflow's progress relates to wall-clock time. None for unknown.
69    pub time_dependence: Option<TimeDependence>,
70}
71
72impl<P, S> DataflowDescription<P, S> {
73    /// Tests if the dataflow refers to a single timestamp, namely
74    /// that `as_of` has a single coordinate and that the `until`
75    /// value corresponds to the `as_of` value plus one, or `as_of`
76    /// is the maximum timestamp and is thus single.
77    pub fn is_single_time(&self) -> bool {
78        // TODO: this would be much easier to check if `until` was a strict lower bound,
79        // and we would be testing that `until == as_of`.
80
81        let until = &self.until;
82
83        // IF `as_of` is not set at all this can't be a single time dataflow.
84        let Some(as_of) = self.as_of.as_ref() else {
85            return false;
86        };
87        // Ensure that as_of <= until.
88        soft_assert_or_log!(
89            timely::PartialOrder::less_equal(as_of, until),
90            "expected empty `as_of ≤ until`, got `{as_of:?} ≰ {until:?}`",
91        );
92        // IF `as_of` is not a single timestamp this can't be a single time dataflow.
93        let Some(as_of) = as_of.as_option() else {
94            return false;
95        };
96        // Ensure that `as_of = MAX` implies `until.is_empty()`.
97        soft_assert_or_log!(
98            as_of != &mz_repr::Timestamp::MAX || until.is_empty(),
99            "expected `until = {{}}` due to `as_of = MAX`, got `until = {until:?}`",
100        );
101        // Note that the `(as_of = MAX, until = {})` case also returns `true`
102        // here (as expected) since we are going to compare two `None` values.
103        as_of.try_step_forward().as_ref() == until.as_option()
104    }
105}
106
107impl DataflowDescription<LirRelationExpr, ()> {
108    /// Check invariants expected to be true about `DataflowDescription`s.
109    pub fn check_invariants(&self) -> Result<(), String> {
110        let mut plans: Vec<_> = self.objects_to_build.iter().map(|o| &o.plan).collect();
111        let mut lir_ids = BTreeSet::new();
112
113        while let Some(plan) = plans.pop() {
114            let lir_id = plan.lir_id;
115            if !lir_ids.insert(lir_id) {
116                return Err(format!(
117                    "duplicate `LirId` in `DataflowDescription`: {lir_id}"
118                ));
119            }
120            plans.extend(plan.node.children());
121        }
122
123        Ok(())
124    }
125}
126
127impl DataflowDescription<OptimizedMirRelationExpr, ()> {
128    /// Imports a previously exported index.
129    ///
130    /// This method makes available an index previously exported as `id`, identified
131    /// to the query by `description` (which names the view the index arranges, and
132    /// the keys by which it is arranged).
133    pub fn import_index(
134        &mut self,
135        id: GlobalId,
136        desc: IndexDesc<MirScalarExpr>,
137        typ: ReprRelationType,
138        monotonic: bool,
139    ) {
140        self.index_imports.insert(
141            id,
142            IndexImport {
143                desc,
144                typ,
145                monotonic,
146                with_snapshot: true,
147            },
148        );
149    }
150
151    /// Imports a source and makes it available as `id`.
152    pub fn import_source(&mut self, id: GlobalId, typ: SqlRelationType, monotonic: bool) {
153        // Import the source with no linear operators applied to it.
154        // They may be populated by whole-dataflow optimization.
155        // Similarly, we require the snapshot by default, though optimization may choose to skip it.
156        self.source_imports.insert(
157            id,
158            SourceImport {
159                desc: SourceInstanceDesc {
160                    storage_metadata: (),
161                    arguments: SourceInstanceArguments { operators: None },
162                    typ,
163                },
164                monotonic,
165                with_snapshot: true,
166                upper: Antichain::new(),
167            },
168        );
169    }
170
171    /// Binds to `id` the relation expression `plan`.
172    pub fn insert_plan(&mut self, id: GlobalId, plan: OptimizedMirRelationExpr) {
173        self.objects_to_build.push(BuildDesc { id, plan });
174    }
175
176    /// Exports as `id` an index described by `description`.
177    ///
178    /// Future uses of `import_index` in other dataflow descriptions may use `id`,
179    /// as long as this dataflow has not been terminated in the meantime.
180    pub fn export_index(
181        &mut self,
182        id: GlobalId,
183        description: IndexDesc<MirScalarExpr>,
184        on_type: ReprRelationType,
185    ) {
186        // We first create a "view" named `id` that ensures that the
187        // data are correctly arranged and available for export.
188        self.insert_plan(
189            id,
190            OptimizedMirRelationExpr::declare_optimized(MirRelationExpr::ArrangeBy {
191                input: Box::new(MirRelationExpr::global_get(
192                    description.on_id,
193                    on_type.clone(),
194                )),
195                keys: vec![description.key.clone()],
196            }),
197        );
198        self.index_exports.insert(id, (description, on_type));
199    }
200
201    /// Exports as `id` a sink described by `description`.
202    pub fn export_sink(&mut self, id: GlobalId, description: ComputeSinkDesc<()>) {
203        self.sink_exports.insert(id, description);
204    }
205
206    /// Returns true iff `id` is already imported.
207    pub fn is_imported(&self, id: &GlobalId) -> bool {
208        self.objects_to_build.iter().any(|bd| &bd.id == id)
209            || self.index_imports.keys().any(|i| i == id)
210            || self.source_imports.keys().any(|i| i == id)
211    }
212
213    /// The number of columns associated with an identifier in the dataflow.
214    pub fn arity_of(&self, id: &GlobalId) -> usize {
215        for (source_id, source_import) in self.source_imports.iter() {
216            let source = &source_import.desc;
217            if source_id == id {
218                return source.typ.arity();
219            }
220        }
221        for IndexImport { desc, typ, .. } in self.index_imports.values() {
222            if &desc.on_id == id {
223                return typ.arity();
224            }
225        }
226        for desc in self.objects_to_build.iter() {
227            if &desc.id == id {
228                return desc.plan.arity();
229            }
230        }
231        panic!("GlobalId {} not found in DataflowDesc", id);
232    }
233
234    /// Calls r and s on any sub-members of those types in self. Halts at the first error return.
235    pub fn visit_children<R, S, E>(&mut self, r: R, s: S) -> Result<(), E>
236    where
237        R: Fn(&mut OptimizedMirRelationExpr) -> Result<(), E>,
238        S: Fn(&mut MirScalarExpr) -> Result<(), E>,
239    {
240        for BuildDesc { plan, .. } in &mut self.objects_to_build {
241            r(plan)?;
242        }
243        for source_import in self.source_imports.values_mut() {
244            let Some(mfp) = source_import.desc.arguments.operators.as_mut() else {
245                continue;
246            };
247            for expr in mfp.expressions.iter_mut() {
248                s(expr)?;
249            }
250            for (_, expr) in mfp.predicates.iter_mut() {
251                s(expr)?;
252            }
253        }
254        Ok(())
255    }
256}
257
258impl<P, S> DataflowDescription<P, S> {
259    /// Creates a new dataflow description with a human-readable name.
260    pub fn new(name: String) -> Self {
261        Self {
262            source_imports: Default::default(),
263            index_imports: Default::default(),
264            objects_to_build: Vec::new(),
265            index_exports: Default::default(),
266            sink_exports: Default::default(),
267            as_of: Default::default(),
268            until: Antichain::new(),
269            initial_storage_as_of: None,
270            refresh_schedule: None,
271            debug_name: name,
272            time_dependence: None,
273        }
274    }
275
276    /// Sets the `as_of` frontier to the supplied argument.
277    ///
278    /// This method allows the dataflow to indicate a frontier up through
279    /// which all times should be advanced. This can be done for at least
280    /// two reasons: 1. correctness and 2. performance.
281    ///
282    /// Correctness may require an `as_of` to ensure that historical detail
283    /// is consolidated at representative times that do not present specific
284    /// detail that is not specifically correct. For example, updates may be
285    /// compacted to times that are no longer the source times, but instead
286    /// some byproduct of when compaction was executed; we should not present
287    /// those specific times as meaningfully different from other equivalent
288    /// times.
289    ///
290    /// Performance may benefit from an aggressive `as_of` as it reduces the
291    /// number of distinct moments at which collections vary. Differential
292    /// dataflow will refresh its outputs at each time its inputs change and
293    /// to moderate that we can minimize the volume of distinct input times
294    /// as much as possible.
295    ///
296    /// Generally, one should consider setting `as_of` at least to the `since`
297    /// frontiers of contributing data sources and as aggressively as the
298    /// computation permits.
299    pub fn set_as_of(&mut self, as_of: Antichain<Timestamp>) {
300        self.as_of = Some(as_of);
301    }
302
303    /// Records the initial `as_of` of the storage collection associated with a materialized view.
304    pub fn set_initial_as_of(&mut self, initial_as_of: Antichain<Timestamp>) {
305        self.initial_storage_as_of = Some(initial_as_of);
306    }
307
308    /// Identifiers of imported objects (indexes and sources).
309    pub fn import_ids(&self) -> impl Iterator<Item = GlobalId> + Clone + '_ {
310        self.imported_index_ids().chain(self.imported_source_ids())
311    }
312
313    /// Identifiers of imported indexes.
314    pub fn imported_index_ids(&self) -> impl Iterator<Item = GlobalId> + Clone + '_ {
315        self.index_imports.keys().copied()
316    }
317
318    /// Identifiers of imported sources.
319    pub fn imported_source_ids(&self) -> impl Iterator<Item = GlobalId> + Clone + '_ {
320        self.source_imports.keys().copied()
321    }
322
323    /// Whether `id` names an import of this dataflow, index or source.
324    pub fn is_import(&self, id: &GlobalId) -> bool {
325        self.index_imports.contains_key(id) || self.source_imports.contains_key(id)
326    }
327
328    /// Identifiers of exported objects (indexes and sinks).
329    pub fn export_ids(&self) -> impl Iterator<Item = GlobalId> + Clone + '_ {
330        self.exported_index_ids().chain(self.exported_sink_ids())
331    }
332
333    /// Identifiers of exported indexes.
334    pub fn exported_index_ids(&self) -> impl Iterator<Item = GlobalId> + Clone + '_ {
335        self.index_exports.keys().copied()
336    }
337
338    /// Identifiers of exported sinks.
339    pub fn exported_sink_ids(&self) -> impl Iterator<Item = GlobalId> + Clone + '_ {
340        self.sink_exports.keys().copied()
341    }
342
343    /// Identifiers of exported persist sinks.
344    pub fn persist_sink_ids(&self) -> impl Iterator<Item = GlobalId> + '_ {
345        self.sink_exports
346            .iter()
347            .filter_map(|(id, desc)| match desc.connection {
348                ComputeSinkConnection::MaterializedView(_) => Some(*id),
349                _ => None,
350            })
351    }
352
353    /// Identifiers of exported subscribe sinks.
354    pub fn subscribe_ids(&self) -> impl Iterator<Item = GlobalId> + '_ {
355        self.sink_exports
356            .iter()
357            .filter_map(|(id, desc)| match desc.connection {
358                ComputeSinkConnection::Subscribe(_) => Some(*id),
359                _ => None,
360            })
361    }
362
363    /// Identifiers of exported copy to sinks.
364    pub fn copy_to_ids(&self) -> impl Iterator<Item = GlobalId> + '_ {
365        self.sink_exports
366            .iter()
367            .filter_map(|(id, desc)| match desc.connection {
368                ComputeSinkConnection::CopyToS3Oneshot(_) => Some(*id),
369                _ => None,
370            })
371    }
372
373    /// Produce a `Display`able value containing the import IDs of this dataflow.
374    pub fn display_import_ids(&self) -> impl fmt::Display + '_ {
375        use mz_ore::str::{bracketed, separated};
376        bracketed("[", "]", separated(", ", self.import_ids()))
377    }
378
379    /// Produce a `Display`able value containing the export IDs of this dataflow.
380    pub fn display_export_ids(&self) -> impl fmt::Display + '_ {
381        use mz_ore::str::{bracketed, separated};
382        bracketed("[", "]", separated(", ", self.export_ids()))
383    }
384
385    /// Whether this dataflow installs transient collections.
386    pub fn is_transient(&self) -> bool {
387        self.export_ids().all(|id| id.is_transient())
388    }
389
390    /// Returns the description of the object to build with the specified
391    /// identifier.
392    ///
393    /// # Panics
394    ///
395    /// Panics if `id` is not present in `objects_to_build` exactly once.
396    pub fn build_desc(&self, id: GlobalId) -> &BuildDesc<P> {
397        let mut builds = self.objects_to_build.iter().filter(|build| build.id == id);
398        let build = builds
399            .next()
400            .unwrap_or_else(|| panic!("object to build id {id} unexpectedly missing"));
401        assert!(builds.next().is_none());
402        build
403    }
404
405    /// Returns the id of the dataflow's sink export.
406    ///
407    /// # Panics
408    ///
409    /// Panics if the dataflow has no sink exports or has more than one.
410    pub fn sink_id(&self) -> GlobalId {
411        let sink_exports = &self.sink_exports;
412        let sink_id = sink_exports.keys().into_element();
413        *sink_id
414    }
415}
416
417impl<P, S> DataflowDescription<P, S>
418where
419    P: CollectionPlan,
420{
421    /// Computes the set of identifiers upon which the specified collection
422    /// identifier depends.
423    ///
424    /// `collection_id` must specify a valid object in `objects_to_build`.
425    ///
426    /// This method includes identifiers for e.g. intermediate views, and should be filtered
427    /// if one only wants sources and indexes.
428    ///
429    /// This method is safe for mutually recursive view definitions.
430    pub fn depends_on(&self, collection_id: GlobalId) -> BTreeSet<GlobalId> {
431        let mut out = BTreeSet::new();
432        self.depends_on_into(collection_id, &mut out);
433        out
434    }
435
436    /// Like `depends_on`, but appends to an existing `BTreeSet`.
437    pub fn depends_on_into(&self, collection_id: GlobalId, out: &mut BTreeSet<GlobalId>) {
438        out.insert(collection_id);
439        if self.source_imports.contains_key(&collection_id) {
440            // The collection is provided by an imported source. Report the
441            // dependency on the source.
442            out.insert(collection_id);
443            return;
444        }
445
446        // NOTE(benesch): we're not smart enough here to know *which* index
447        // for the collection will be used, if one exists, so we have to report
448        // the dependency on all of them.
449        let mut found_index = false;
450        for (index_id, IndexImport { desc, .. }) in &self.index_imports {
451            if desc.on_id == collection_id {
452                // The collection is provided by an imported index. Report the
453                // dependency on the index.
454                out.insert(*index_id);
455                found_index = true;
456            }
457        }
458        if found_index {
459            return;
460        }
461
462        // The collection is not provided by a source or imported index.
463        // It must be a collection whose plan we have handy. Recurse.
464        let build = self.build_desc(collection_id);
465        for id in build.plan.depends_on() {
466            if !out.contains(&id) {
467                self.depends_on_into(id, out)
468            }
469        }
470    }
471
472    /// Computes the set of imports upon which the specified collection depends.
473    ///
474    /// This method behaves like `depends_on` but filters out internal dependencies that are not
475    /// included in the dataflow imports.
476    pub fn depends_on_imports(&self, collection_id: GlobalId) -> BTreeSet<GlobalId> {
477        let deps = self.depends_on(collection_id);
478        deps.into_iter().filter(|id| self.is_import(id)).collect()
479    }
480
481    /// Computes the set of imports the dataflow's exports read, meaning the imports reachable from
482    /// an index export's `on_id` or a sink export's `from`.
483    ///
484    /// A description that has been through the optimizer answers [`Self::import_ids`] here, because
485    /// the optimizer prunes the import list to what the exports read. The two come apart while a
486    /// description is still being assembled, and this is what the prune and the assertion guarding
487    /// it are both defined in terms of. The answer covers the dataflow as a whole, so an import only
488    /// one export reads is still reported, and a dataflow with no exports reports none.
489    ///
490    /// NOTE: On the index side this over-approximates. [`Self::depends_on`] cannot tell which index
491    /// on a collection a plan will use, so reaching a collection reports every index imported on it.
492    /// Pruning index imports needs the exact usage information the MIR pipeline collects, not this.
493    ///
494    /// Panics for an export naming a collection that is neither an import nor built exactly once
495    /// here, which is [`Self::depends_on`]'s precondition on its argument. Rendering resolves the
496    /// same ids, so a description that trips this does not survive being built either.
497    pub fn used_import_ids(&self) -> BTreeSet<GlobalId> {
498        let mut deps = BTreeSet::new();
499        for (index_desc, _typ) in self.index_exports.values() {
500            self.depends_on_into(index_desc.on_id, &mut deps);
501        }
502        for sink_desc in self.sink_exports.values() {
503            self.depends_on_into(sink_desc.from, &mut deps);
504        }
505        deps.retain(|id| self.is_import(id));
506        deps
507    }
508}
509
510impl<S> DataflowDescription<RenderPlan, S>
511where
512    S: Clone + PartialEq,
513{
514    /// Determine if a dataflow description is compatible with this dataflow description.
515    ///
516    /// Compatible dataflows have structurally equal exports, imports, and objects to build. The
517    /// `as_of` of the receiver has to be less equal the `other` `as_of`.
518    ///
519    /// Note that this method performs normalization as part of the structural equality checking,
520    /// which involves cloning both `self` and `other`. It is therefore relatively expensive and
521    /// should only be used on cold code paths.
522    ///
523    // TODO: The semantics of this function are only useful for command reconciliation at the moment.
524    pub fn compatible_with(&self, other: &Self) -> bool {
525        let old = self.as_comparable();
526        let new = other.as_comparable();
527
528        let equality = old.index_exports == new.index_exports
529            && old.sink_exports == new.sink_exports
530            && old.objects_to_build == new.objects_to_build
531            && old.index_imports == new.index_imports
532            && old.source_imports == new.source_imports
533            && old.time_dependence == new.time_dependence;
534
535        let partial = if let (Some(old_as_of), Some(new_as_of)) = (&old.as_of, &new.as_of) {
536            timely::PartialOrder::less_equal(old_as_of, new_as_of)
537        } else {
538            false
539        };
540
541        equality && partial
542    }
543
544    /// Returns a `DataflowDescription` that has the same structure as `self` and can be
545    /// structurally compared to other `DataflowDescription`s.
546    ///
547    /// The function normalizes several properties. It replaces transient `GlobalId`s
548    /// that are only used internally (i.e. not imported nor exported) with consecutive IDs
549    /// starting from `t1`. It replaces the source import's `upper` by a dummy value.
550    fn as_comparable(&self) -> Self {
551        let external_ids: BTreeSet<_> = self.import_ids().chain(self.export_ids()).collect();
552
553        let mut id_counter = 0;
554        let mut replacements = BTreeMap::new();
555
556        let mut maybe_replace = |id: GlobalId| {
557            if id.is_transient() && !external_ids.contains(&id) {
558                *replacements.entry(id).or_insert_with(|| {
559                    id_counter += 1;
560                    GlobalId::Transient(id_counter)
561                })
562            } else {
563                id
564            }
565        };
566
567        let mut source_imports = self.source_imports.clone();
568        for import in source_imports.values_mut() {
569            import.upper = Antichain::new();
570        }
571
572        let mut objects_to_build = self.objects_to_build.clone();
573        for object in &mut objects_to_build {
574            object.id = maybe_replace(object.id);
575            object.plan.replace_ids(&mut maybe_replace);
576        }
577
578        let mut index_exports = self.index_exports.clone();
579        for (desc, _typ) in index_exports.values_mut() {
580            desc.on_id = maybe_replace(desc.on_id);
581        }
582
583        let mut sink_exports = self.sink_exports.clone();
584        for desc in sink_exports.values_mut() {
585            desc.from = maybe_replace(desc.from);
586        }
587
588        DataflowDescription {
589            source_imports,
590            index_imports: self.index_imports.clone(),
591            objects_to_build,
592            index_exports,
593            sink_exports,
594            as_of: self.as_of.clone(),
595            until: self.until.clone(),
596            initial_storage_as_of: self.initial_storage_as_of.clone(),
597            refresh_schedule: self.refresh_schedule.clone(),
598            debug_name: self.debug_name.clone(),
599            time_dependence: self.time_dependence.clone(),
600        }
601    }
602}
603
604/// A commonly used name for dataflows contain MIR expressions.
605pub type DataflowDesc = DataflowDescription<OptimizedMirRelationExpr, ()>;
606
607/// An index storing processed updates so they can be queried
608/// or reused in other computations
609#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize, Hash)]
610pub struct IndexDesc<E> {
611    /// Identity of the collection the index is on.
612    pub on_id: GlobalId,
613    /// Expressions to be arranged, in order of decreasing primacy.
614    pub key: Vec<E>,
615}
616
617impl IndexDesc<MirScalarExpr> {
618    /// Translate an index description from MIR to LIR.
619    pub fn as_lir(&self) -> IndexDesc<LirScalarExpr> {
620        let on_id = self.on_id.clone();
621        let key = lses_from_mses(&self.key);
622
623        IndexDesc { on_id, key }
624    }
625}
626
627/// Information about an imported index, and how it will be used by the dataflow.
628#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
629pub struct IndexImport {
630    /// Description of index.
631    pub desc: IndexDesc<MirScalarExpr>,
632    /// Schema and keys of the object the index is on.
633    pub typ: ReprRelationType,
634    /// Whether the index will supply monotonic data.
635    pub monotonic: bool,
636    /// Whether this import must include the snapshot data.
637    pub with_snapshot: bool,
638}
639
640/// Information about an imported source, and how it will be used by the dataflow.
641#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
642pub struct SourceImport<S: 'static = ()> {
643    /// Description of the source instance to import.
644    pub desc: SourceInstanceDesc<S>,
645    /// Whether the source will supply monotonic data.
646    pub monotonic: bool,
647    /// Whether this import must include the snapshot data.
648    pub with_snapshot: bool,
649    /// The initial known upper frontier for the source.
650    pub upper: Antichain<Timestamp>,
651}
652
653/// An association of a global identifier to an expression.
654#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
655pub struct BuildDesc<P> {
656    /// TODO(database-issues#7533): Add documentation.
657    pub id: GlobalId,
658    /// TODO(database-issues#7533): Add documentation.
659    pub plan: P,
660}
661
662#[cfg(test)]
663mod tests {
664    use mz_expr::{AccessStrategy, Id, MirRelationExpr};
665    use mz_repr::{RelationDesc, ReprRelationType, ReprScalarType, SqlRelationType};
666
667    use crate::sinks::{ComputeSinkConnection, ComputeSinkDesc, SubscribeSinkConnection};
668    use crate::sources::{SourceInstanceArguments, SourceInstanceDesc};
669
670    use super::*;
671
672    const READ: GlobalId = GlobalId::User(1);
673    const UNREAD: GlobalId = GlobalId::User(2);
674    const VIEW: GlobalId = GlobalId::Transient(1);
675    const SINK: GlobalId = GlobalId::Transient(2);
676    const INDEX: GlobalId = GlobalId::Transient(3);
677
678    fn typ() -> ReprRelationType {
679        ReprRelationType::new(vec![ReprScalarType::Int64.nullable(false)])
680    }
681
682    /// A dataflow importing `READ` and `UNREAD`, building `VIEW` from `plan`, and exporting a
683    /// subscribe sink over it.
684    fn dataflow(plan: MirRelationExpr) -> DataflowDesc {
685        let source_import = || SourceImport {
686            desc: SourceInstanceDesc {
687                arguments: SourceInstanceArguments { operators: None },
688                storage_metadata: (),
689                typ: SqlRelationType::from_repr(&typ()),
690            },
691            monotonic: false,
692            with_snapshot: true,
693            upper: Antichain::from_elem(Timestamp::MIN),
694        };
695
696        let mut df = DataflowDesc::new("test".to_string());
697        df.source_imports.insert(READ, source_import());
698        df.source_imports.insert(UNREAD, source_import());
699        df.objects_to_build.push(BuildDesc {
700            id: VIEW,
701            plan: OptimizedMirRelationExpr::declare_optimized(plan),
702        });
703        df.sink_exports.insert(
704            SINK,
705            ComputeSinkDesc {
706                from: VIEW,
707                from_desc: RelationDesc::new(SqlRelationType::from_repr(&typ()), ["c"]),
708                connection: ComputeSinkConnection::Subscribe(SubscribeSinkConnection {
709                    output: Vec::new(),
710                }),
711                with_snapshot: true,
712                up_to: Antichain::new(),
713                non_null_assertions: Vec::new(),
714                refresh_schedule: None,
715            },
716        );
717        df
718    }
719
720    #[mz_ore::test]
721    fn used_import_ids_reports_only_read_imports() {
722        let df = dataflow(MirRelationExpr::Get {
723            id: Id::Global(READ),
724            typ: typ(),
725            access_strategy: AccessStrategy::Persist,
726        });
727
728        assert_eq!(df.used_import_ids(), BTreeSet::from([READ]));
729    }
730
731    /// An export the optimizer folded to a constant reads nothing, even though the imports it was
732    /// folded from are still there. This is the shape that must not be reported as reading them.
733    #[mz_ore::test]
734    fn used_import_ids_is_empty_for_a_constant_export() {
735        let df = dataflow(MirRelationExpr::Constant {
736            rows: Ok(Vec::new()),
737            typ: typ(),
738        });
739
740        assert_eq!(df.used_import_ids(), BTreeSet::new());
741    }
742
743    /// Index exports are walked from the collection they are on, just like sink exports are from
744    /// the collection they are from.
745    #[mz_ore::test]
746    fn used_import_ids_covers_index_exports() {
747        let mut df = dataflow(MirRelationExpr::Constant {
748            rows: Ok(Vec::new()),
749            typ: typ(),
750        });
751        let other_view = GlobalId::Transient(4);
752        df.objects_to_build.push(BuildDesc {
753            id: other_view,
754            plan: OptimizedMirRelationExpr::declare_optimized(MirRelationExpr::Get {
755                id: Id::Global(UNREAD),
756                typ: typ(),
757                access_strategy: AccessStrategy::Persist,
758            }),
759        });
760        df.index_exports.insert(
761            INDEX,
762            (
763                IndexDesc {
764                    on_id: other_view,
765                    key: Vec::new(),
766                },
767                typ(),
768            ),
769        );
770
771        assert_eq!(df.used_import_ids(), BTreeSet::from([UNREAD]));
772    }
773
774    /// An imported index is reached through the collection it is on, and reported by the id of the
775    /// index rather than that of the collection.
776    #[mz_ore::test]
777    fn used_import_ids_reports_imported_indexes_by_index_id() {
778        let indexed_view = GlobalId::User(3);
779        let imported_index = GlobalId::User(4);
780
781        let mut df = dataflow(MirRelationExpr::Get {
782            id: Id::Global(indexed_view),
783            typ: typ(),
784            access_strategy: AccessStrategy::Index(Vec::new()),
785        });
786        df.index_imports.insert(
787            imported_index,
788            IndexImport {
789                desc: IndexDesc {
790                    on_id: indexed_view,
791                    key: Vec::new(),
792                },
793                typ: typ(),
794                monotonic: false,
795                with_snapshot: true,
796            },
797        );
798
799        assert_eq!(df.used_import_ids(), BTreeSet::from([imported_index]));
800    }
801}
mz_compute_types/dataflows.rs

mz_compute_types/
dataflows.rs
