mz_storage/source/
sql_server.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//! Code to render the ingestion dataflow of a [`SqlServerSourceConnection`].
11
12use std::collections::BTreeMap;
13use std::future::Future;
14use std::rc::Rc;
15use std::sync::Arc;
16
17use differential_dataflow::AsCollection;
18use itertools::Itertools;
19use mz_ore::cast::CastFrom;
20use mz_ore::error::ErrorExt;
21use mz_repr::{Diff, GlobalId};
22use mz_sql_server_util::SqlServerError;
23use mz_sql_server_util::cdc::Lsn;
24use mz_sql_server_util::desc::{SqlServerRowDecoder, SqlServerTableDesc};
25use mz_storage_types::errors::{DataflowError, SourceError, SourceErrorDetails};
26use mz_storage_types::sources::{
27    SourceExport, SourceExportDetails, SourceTimestamp, SqlServerSourceConnection,
28};
29use mz_timely_util::builder_async::PressOnDropButton;
30use timely::container::CapacityContainerBuilder;
31use timely::dataflow::operators::Concat;
32use timely::dataflow::operators::core::Partition;
33use timely::dataflow::operators::vec::{Map, ToStream};
34use timely::dataflow::{Scope, StreamVec};
35use timely::progress::{Antichain, Timestamp};
36
37use crate::healthcheck::{HealthStatusMessage, HealthStatusUpdate, StatusNamespace};
38use crate::source::RawSourceCreationConfig;
39use crate::source::types::{Probe, SourceMessage, SourceRender, StackedCollection};
40
41mod progress;
42mod replication;
43
44#[derive(Debug, Clone)]
45struct SourceOutputInfo {
46    /// Name of the capture instance in the upstream SQL Server DB.
47    capture_instance: Arc<str>,
48    /// Description of the upstream table.
49    #[allow(dead_code)]
50    upstream_desc: Arc<SqlServerTableDesc>,
51    /// Type that can decode (and map) SQL Server rows into Materialize rows.
52    decoder: Arc<SqlServerRowDecoder>,
53    /// Upper to resume replication from.
54    resume_upper: Antichain<Lsn>,
55    /// An index to split the timely stream.
56    partition_index: u64,
57    /// The basis for the resumption LSN when snapshotting.
58    initial_lsn: Lsn,
59}
60
61impl SourceOutputInfo {
62    /// The [`Lsn`] this output resumes reading from, or the provided fallback [`Lsn`].
63    ///
64    /// Panics if `resume_upper` is empty, which would mean the output has no
65    /// resumption point at all.
66    fn resume_lsn_or(&self, fallback: Lsn) -> Lsn {
67        match self.resume_upper.as_option() {
68            Some(lsn) if *lsn != Lsn::minimum() => *lsn,
69            Some(_) => fallback,
70            None => panic!("resume_upper has at least one value"),
71        }
72    }
73}
74
75#[derive(Debug, Clone, thiserror::Error)]
76pub enum ReplicationError {
77    #[error(transparent)]
78    Transient(#[from] Rc<TransientError>),
79    #[error(transparent)]
80    DefiniteError(#[from] Rc<DefiniteError>),
81}
82
83#[derive(Debug, thiserror::Error)]
84pub enum TransientError {
85    #[error("stream ended prematurely")]
86    ReplicationEOF,
87    #[error(transparent)]
88    SqlServer(#[from] SqlServerError),
89    #[error(transparent)]
90    Generic(#[from] anyhow::Error),
91}
92
93#[derive(Debug, Clone, thiserror::Error)]
94pub enum DefiniteError {
95    #[error("unable to decode: {0}")]
96    ValueDecodeError(String),
97    #[error("failed to decode row: {0}")]
98    Decoding(String),
99    #[error("programming error: {0}")]
100    ProgrammingError(String),
101    #[error("Restore history id changed from {0:?} to {1:?}")]
102    RestoreHistoryChanged(Option<i32>, Option<i32>),
103    #[error("Incompatible schema change for table {0} capture instance {1}")]
104    IncompatibleSchemaChange(String, String),
105}
106
107impl From<DefiniteError> for DataflowError {
108    fn from(val: DefiniteError) -> Self {
109        let msg = val.to_string().into();
110        DataflowError::SourceError(Box::new(SourceError {
111            error: SourceErrorDetails::Other(msg),
112        }))
113    }
114}
115
116impl SourceRender for SqlServerSourceConnection {
117    type Time = Lsn;
118
119    const STATUS_NAMESPACE: StatusNamespace = StatusNamespace::SqlServer;
120
121    fn render<'scope>(
122        self,
123        scope: Scope<'scope, Lsn>,
124        config: &RawSourceCreationConfig,
125        resume_uppers: impl futures::Stream<Item = Antichain<Lsn>> + 'static,
126        _start_signal: impl Future<Output = ()> + 'static,
127    ) -> (
128        // Timely Collection for each Source Export defined in the provided `config`.
129        BTreeMap<GlobalId, StackedCollection<'scope, Lsn, Result<SourceMessage, DataflowError>>>,
130        StreamVec<'scope, Lsn, HealthStatusMessage>,
131        StreamVec<'scope, Lsn, Probe<Lsn>>,
132        Vec<PressOnDropButton>,
133    ) {
134        // Collect the source outputs that we will be exporting.
135        let mut source_outputs = BTreeMap::new();
136        for (idx, (id, export)) in config.source_exports.iter().enumerate() {
137            let SourceExport {
138                details,
139                storage_metadata,
140                data_config: _,
141            } = export;
142
143            let details = match details {
144                SourceExportDetails::SqlServer(details) => details,
145                // This is an export that doesn't need any data output to it.
146                SourceExportDetails::None => continue,
147                other => unreachable!("unexpected source export details: {other:?}"),
148            };
149
150            let decoder = details
151                .table
152                .decoder(&storage_metadata.relation_desc)
153                .expect("TODO handle errors");
154            let upstream_desc = Arc::new(details.table.clone());
155            let resume_upper = config
156                .source_resume_uppers
157                .get(id)
158                .expect("missing resume upper")
159                .iter()
160                .map(Lsn::decode_row);
161
162            let output_info = SourceOutputInfo {
163                capture_instance: Arc::clone(&details.capture_instance),
164                upstream_desc,
165                decoder: Arc::new(decoder),
166                resume_upper: Antichain::from_iter(resume_upper),
167                partition_index: u64::cast_from(idx),
168                initial_lsn: details.initial_lsn,
169            };
170            source_outputs.insert(*id, output_info);
171        }
172
173        let metrics = config
174            .metrics
175            .get_sql_server_source_metrics(config.id, config.worker_id);
176
177        let (repl_updates, repl_errs, repl_token) = replication::render(
178            scope.clone(),
179            config.clone(),
180            source_outputs.clone(),
181            self.clone(),
182            metrics,
183        );
184
185        let (progress_errs, progress_probes, progress_token) = progress::render(
186            scope.clone(),
187            config.clone(),
188            self.connection.clone(),
189            source_outputs.clone(),
190            resume_uppers,
191            self.extras.clone(),
192        );
193
194        let partition_count = u64::cast_from(config.source_exports.len());
195        let data_streams: Vec<_> = repl_updates
196            .inner
197            .partition::<CapacityContainerBuilder<_>, _, _>(
198                partition_count,
199                move |((partition_idx, data), time, diff): (
200                    (u64, Result<SourceMessage, DataflowError>),
201                    Lsn,
202                    Diff,
203                )| { (partition_idx, (data, time, diff)) },
204            );
205        let mut data_collections = BTreeMap::new();
206        for (id, data_stream) in config.source_exports.keys().zip_eq(data_streams) {
207            data_collections.insert(*id, data_stream.as_collection());
208        }
209
210        let export_ids = config.source_exports.keys().copied();
211        let health_init = export_ids
212            .map(Some)
213            .chain(std::iter::once(None))
214            .map(|id| HealthStatusMessage {
215                id,
216                namespace: Self::STATUS_NAMESPACE,
217                update: HealthStatusUpdate::Running,
218            })
219            .collect::<Vec<_>>()
220            .to_stream(scope);
221
222        let health_errs = repl_errs.concat(progress_errs).map(move |err| {
223            // This update will cause the dataflow to restart
224            let err_string = err.display_with_causes().to_string();
225            let update = HealthStatusUpdate::halting(err_string, None);
226            // TODO(sql_server2): If the error has anything to do with SSH
227            // connections we should use the SSH status namespace.
228            let namespace = Self::STATUS_NAMESPACE;
229
230            HealthStatusMessage {
231                id: None,
232                namespace: namespace.clone(),
233                update,
234            }
235        });
236        let health = health_init.concat(health_errs);
237
238        (
239            data_collections,
240            health,
241            progress_probes,
242            vec![repl_token, progress_token],
243        )
244    }
245}
mz_storage/source/sql_server.rs

mz_storage/source/
sql_server.rs
