mz_adapter/coord/
read_then_write.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//! Coordinator-side support machinery for (frontend) read-then write.
11
12use std::collections::BTreeSet;
13
14use mz_catalog::memory::objects::CatalogItem;
15use mz_repr::CatalogItemId;
16use mz_sql::catalog::CatalogItemType;
17
18use crate::catalog::Catalog;
19use crate::error::AdapterError;
20
21/// Adds `id` to the worklist the first time it is seen, enforcing the
22/// dependency bound.
23///
24/// Deduping at enqueue time keeps `seen` and `stack` proportional to the number
25/// of distinct objects, not the number of dependency edges. A diamond-shaped
26/// graph is validated once per object.
27fn enqueue(
28    seen: &mut BTreeSet<CatalogItemId>,
29    stack: &mut Vec<CatalogItemId>,
30    id: CatalogItemId,
31    max_rw_dependencies: usize,
32) -> Result<(), AdapterError> {
33    if seen.insert(id) {
34        if seen.len() > max_rw_dependencies {
35            return Err(AdapterError::ReadThenWriteDependencyLimitExceeded {
36                max_rw_dependencies,
37            });
38        }
39        stack.push(id);
40    }
41    Ok(())
42}
43
44/// Validates that all dependencies are valid for read-then-write operations.
45///
46/// Ensures all objects the selection transitively depends on (seeded by `ids`) are valid for
47/// `ReadThenWrite` operations:
48///
49/// - They do not refer to any objects whose notion of time moves differently than that of
50///   user tables. This limitation is meant to ensure no writes occur between this read and the
51///   subsequent write.
52/// - They do not use mz_now(), whose time produced during read will differ from the write
53///   timestamp.
54///
55/// The first invalid or temporal dependency encountered short-circuits with the corresponding
56/// error. Traversal is bounded at `max_rw_dependencies` distinct objects, returning
57/// [`AdapterError::ReadThenWriteDependencyLimitExceeded`] if exceeded.
58pub(crate) fn validate_read_then_write_dependencies(
59    catalog: &Catalog,
60    ids: impl IntoIterator<Item = CatalogItemId>,
61    max_rw_dependencies: usize,
62) -> Result<(), AdapterError> {
63    use CatalogItemType::*;
64    use mz_catalog::memory::objects;
65
66    // Iterative worklist rather than recursion. Dependency chains are user
67    // controlled and can be arbitrarily deep (e.g. a long chain of stacked
68    // views), so recursing risks a stack overflow on the coordinator thread.
69    let mut seen = BTreeSet::new();
70    let mut stack = Vec::new();
71    for id in ids {
72        enqueue(&mut seen, &mut stack, id, max_rw_dependencies)?;
73    }
74    while let Some(id) = stack.pop() {
75        let mut ids_to_check = Vec::new();
76        let valid = match catalog.try_get_entry(&id) {
77            Some(entry) => {
78                if let CatalogItem::View(objects::View {
79                    locally_optimized_expr: optimized_expr,
80                    ..
81                })
82                | CatalogItem::MaterializedView(objects::MaterializedView {
83                    locally_optimized_expr: optimized_expr,
84                    ..
85                }) = entry.item()
86                {
87                    if optimized_expr.contains_temporal() {
88                        return Err(AdapterError::Unsupported(
89                            "calls to mz_now in write statements",
90                        ));
91                    }
92                }
93                match entry.item().typ() {
94                    typ @ (Func | View | MaterializedView) => {
95                        ids_to_check.extend(entry.uses());
96                        let valid_id = id.is_user() || matches!(typ, Func);
97                        valid_id
98                    }
99                    Source | Secret | Connection => false,
100                    // Cannot select from sinks or indexes.
101                    Sink | Index => unreachable!(),
102                    Table => {
103                        if !id.is_user() {
104                            // We can't read from non-user tables
105                            false
106                        } else {
107                            // We can't read from tables that are source-exports
108                            entry.source_export_details().is_none()
109                        }
110                    }
111                    Type => true,
112                }
113            }
114            None => false,
115        };
116        if !valid {
117            let (object_name, object_type) = match catalog.try_get_entry(&id) {
118                Some(entry) => {
119                    let object_name = catalog.resolve_full_name(entry.name(), None).to_string();
120                    let object_type = match entry.item().typ() {
121                        // We only need the disallowed types here; the allowed types are handled above.
122                        Source => "source",
123                        Secret => "secret",
124                        Connection => "connection",
125                        Table => {
126                            if !id.is_user() {
127                                "system table"
128                            } else {
129                                "source-export table"
130                            }
131                        }
132                        View => "system view",
133                        MaterializedView => "system materialized view",
134                        _ => "invalid dependency",
135                    };
136                    (object_name, object_type.to_string())
137                }
138                None => (id.to_string(), "unknown".to_string()),
139            };
140            return Err(AdapterError::InvalidTableMutationSelection {
141                object_name,
142                object_type,
143            });
144        }
145        for dep in ids_to_check {
146            enqueue(&mut seen, &mut stack, dep, max_rw_dependencies)?;
147        }
148    }
149    Ok(())
150}
mz_adapter/coord/read_then_write.rs

mz_adapter/coord/
read_then_write.rs
