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            hint: None,
113        }))
114    }
115}
116
117impl SourceRender for SqlServerSourceConnection {
118    type Time = Lsn;
119
120    const STATUS_NAMESPACE: StatusNamespace = StatusNamespace::SqlServer;
121
122    fn render<'scope>(
123        self,
124        scope: Scope<'scope, Lsn>,
125        config: &RawSourceCreationConfig,
126        resume_uppers: impl futures::Stream<Item = Antichain<Lsn>> + 'static,
127        _start_signal: impl Future<Output = ()> + 'static,
128    ) -> (
129        // Timely Collection for each Source Export defined in the provided `config`.
130        BTreeMap<GlobalId, StackedCollection<'scope, Lsn, Result<SourceMessage, DataflowError>>>,
131        StreamVec<'scope, Lsn, HealthStatusMessage>,
132        StreamVec<'scope, Lsn, Probe<Lsn>>,
133        Vec<PressOnDropButton>,
134    ) {
135        // Collect the source outputs that we will be exporting.
136        let mut source_outputs = BTreeMap::new();
137        for (idx, (id, export)) in config.source_exports.iter().enumerate() {
138            let SourceExport {
139                details,
140                storage_metadata,
141                data_config: _,
142            } = export;
143
144            let details = match details {
145                SourceExportDetails::SqlServer(details) => details,
146                // This is an export that doesn't need any data output to it.
147                SourceExportDetails::None => continue,
148                other => unreachable!("unexpected source export details: {other:?}"),
149            };
150
151            let decoder = details
152                .table
153                .decoder(&storage_metadata.relation_desc)
154                .expect("TODO handle errors");
155            let upstream_desc = Arc::new(details.table.clone());
156            let resume_upper = config
157                .source_resume_uppers
158                .get(id)
159                .expect("missing resume upper")
160                .iter()
161                .map(Lsn::decode_row);
162
163            let output_info = SourceOutputInfo {
164                capture_instance: Arc::clone(&details.capture_instance),
165                upstream_desc,
166                decoder: Arc::new(decoder),
167                resume_upper: Antichain::from_iter(resume_upper),
168                partition_index: u64::cast_from(idx),
169                initial_lsn: details.initial_lsn,
170            };
171            source_outputs.insert(*id, output_info);
172        }
173
174        let metrics = config
175            .metrics
176            .get_sql_server_source_metrics(config.id, config.worker_id);
177
178        let (repl_updates, repl_errs, repl_token) = replication::render(
179            scope.clone(),
180            config.clone(),
181            source_outputs.clone(),
182            self.clone(),
183            metrics,
184        );
185
186        let (progress_errs, progress_probes, progress_token) = progress::render(
187            scope.clone(),
188            config.clone(),
189            self.connection.clone(),
190            source_outputs.clone(),
191            resume_uppers,
192            self.extras.clone(),
193        );
194
195        let partition_count = u64::cast_from(config.source_exports.len());
196        let data_streams: Vec<_> = repl_updates
197            .inner
198            .partition::<CapacityContainerBuilder<_>, _, _>(
199                partition_count,
200                move |((partition_idx, data), time, diff): (
201                    (u64, Result<SourceMessage, DataflowError>),
202                    Lsn,
203                    Diff,
204                )| { (partition_idx, (data, time, diff)) },
205            );
206        let mut data_collections = BTreeMap::new();
207        for (id, data_stream) in config.source_exports.keys().zip_eq(data_streams) {
208            data_collections.insert(*id, data_stream.as_collection());
209        }
210
211        let export_ids = config.source_exports.keys().copied();
212        let health_init = export_ids
213            .map(Some)
214            .chain(std::iter::once(None))
215            .map(|id| HealthStatusMessage {
216                id,
217                namespace: Self::STATUS_NAMESPACE,
218                update: HealthStatusUpdate::Running,
219            })
220            .collect::<Vec<_>>()
221            .to_stream(scope);
222
223        let health_errs = repl_errs.concat(progress_errs).map(move |err| {
224            // This update will cause the dataflow to restart
225            let err_string = err.display_with_causes().to_string();
226            let update = HealthStatusUpdate::halting(err_string, None);
227            // TODO(sql_server2): If the error has anything to do with SSH
228            // connections we should use the SSH status namespace.
229            let namespace = Self::STATUS_NAMESPACE;
230
231            HealthStatusMessage {
232                id: None,
233                namespace: namespace.clone(),
234                update,
235            }
236        });
237        let health = health_init.concat(health_errs);
238
239        (
240            data_collections,
241            health,
242            progress_probes,
243            vec![repl_token, progress_token],
244        )
245    }
246}
mz_storage/source/sql_server.rs

mz_storage/source/
sql_server.rs
