mz_adapter/coord/sequencer/inner/
explain_timestamp.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
10use chrono::{DateTime, Utc};
11use itertools::Itertools;
12use mz_adapter_types::connection::ConnectionId;
13use mz_controller_types::ClusterId;
14
15use crate::coord::Coordinator;
16use crate::coord::timestamp_selection::{TimestampDetermination, TimestampSource};
17use crate::{AdapterError, CollectionIdBundle, TimestampExplanation};
18
19impl Coordinator {
20    /// Explains `determination`, the timestamp chosen for a statement that reads `id_bundle`.
21    ///
22    /// Fails with [`AdapterError::ConcurrentDependencyDrop`] if a collection in `id_bundle` no
23    /// longer exists. Callers determine the timestamp and its read holds before sending
24    /// [`crate::command::Command::ExplainTimestamp`], and a DROP can land in between: read holds
25    /// hold back compaction, not drops.
26    pub(crate) fn explain_timestamp(
27        &self,
28        conn_id: &ConnectionId,
29        session_wall_time: DateTime<Utc>,
30        cluster_id: ClusterId,
31        id_bundle: &CollectionIdBundle,
32        determination: TimestampDetermination,
33    ) -> Result<TimestampExplanation, AdapterError> {
34        let mut sources = Vec::new();
35        {
36            let storage_ids = id_bundle.storage_ids.iter().cloned().collect_vec();
37            let frontiers = self
38                .controller
39                .storage
40                .collections_frontiers(storage_ids)
41                .map_err(AdapterError::concurrent_dependency_drop_from_collection_missing)?;
42
43            for (id, since, upper) in frontiers {
44                let name = self
45                    .catalog()
46                    .try_get_entry_by_global_id(&id)
47                    .map(|item| item.name())
48                    .map(|name| {
49                        self.catalog()
50                            .resolve_full_name(name, Some(conn_id))
51                            .to_string()
52                    })
53                    .unwrap_or_else(|| id.to_string());
54                sources.push(TimestampSource {
55                    name: format!("{name} ({id}, storage)"),
56                    read_frontier: since.elements().to_vec(),
57                    write_frontier: upper.elements().to_vec(),
58                });
59            }
60        }
61        {
62            if let Some(compute_ids) = id_bundle.compute_ids.get(&cluster_id) {
63                let catalog = self.catalog();
64                for id in compute_ids {
65                    let frontiers = self
66                        .controller
67                        .compute
68                        .collection_frontiers(*id, Some(cluster_id))
69                        .map_err(
70                            AdapterError::concurrent_dependency_drop_from_compute_lookup_error,
71                        )?;
72                    let name = catalog
73                        .try_get_entry_by_global_id(id)
74                        .map(|item| item.name())
75                        .map(|name| catalog.resolve_full_name(name, Some(conn_id)).to_string())
76                        .unwrap_or_else(|| id.to_string());
77                    sources.push(TimestampSource {
78                        name: format!("{name} ({id}, compute)"),
79                        read_frontier: frontiers.read_frontier.to_vec(),
80                        write_frontier: frontiers.write_frontier.to_vec(),
81                    });
82                }
83            }
84        }
85        let respond_immediately = determination.respond_immediately();
86        Ok(TimestampExplanation {
87            determination,
88            sources,
89            session_wall_time,
90            respond_immediately,
91        })
92    }
93}
mz_adapter/coord/sequencer/inner/explain_timestamp.rs

mz_adapter/coord/sequencer/inner/
explain_timestamp.rs
