Skip to main content

mz_sql_server_util/
desc.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//! Metadata about tables, columns, and other objects from SQL Server.
11//!
12//! ### Tables
13//!
14//! When creating a SQL Server source we will query system tables from the
15//! upstream instance to get a [`SqlServerTableRaw`]. From this raw information
16//! we create a [`SqlServerTableDesc`] which describes how the upstream table
17//! will get represented in Materialize.
18//!
19//! ### Rows
20//!
21//! With a [`SqlServerTableDesc`] and an [`mz_repr::RelationDesc`] we can
22//! create a [`SqlServerRowDecoder`] which will be used when running a source
23//! to efficiently decode [`tiberius::Row`]s into [`mz_repr::Row`]s.
24
25use base64::Engine;
26use chrono::{NaiveDateTime, SubsecRound};
27use dec::OrderedDecimal;
28use mz_ore::cast::CastFrom;
29use mz_proto::{IntoRustIfSome, ProtoType, RustType};
30use mz_repr::adt::numeric::{Numeric, NumericMaxScale};
31use mz_repr::adt::timestamp::{CheckedTimestamp, TimestampPrecision};
32use mz_repr::adt::varchar::VarCharMaxLength;
33use mz_repr::{Datum, RelationDesc, Row, RowArena, SqlColumnType, SqlScalarType};
34#[cfg(any(test, feature = "proptest"))]
35use proptest_derive::Arbitrary;
36use serde::{Deserialize, Serialize};
37
38use std::collections::BTreeSet;
39use std::sync::Arc;
40
41use crate::desc::proto_sql_server_table_constraint::ConstraintType;
42use crate::{SqlServerDecodeError, SqlServerError};
43
44include!(concat!(env!("OUT_DIR"), "/mz_sql_server_util.rs"));
45
46/// Materialize compatible description of a table in Microsoft SQL Server.
47///
48/// See [`SqlServerTableRaw`] for the raw information we read from the upstream
49/// system.
50///
51/// Note: We map a [`SqlServerTableDesc`] to a Materialize [`RelationDesc`] as
52/// part of purification. Specifically we use this description to generate a
53/// SQL statement for subsource and it's the _parsing of that statement_ which
54/// actually generates a [`RelationDesc`].
55#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
56#[cfg_attr(any(test, feature = "proptest"), derive(Arbitrary))]
57pub struct SqlServerTableDesc {
58    /// Name of the schema that the table belongs to.
59    pub schema_name: Arc<str>,
60    /// Name of the table.
61    pub name: Arc<str>,
62    /// Columns for the table.
63    pub columns: Box<[SqlServerColumnDesc]>,
64    /// Constraints for the table.
65    pub constraints: Vec<SqlServerTableConstraint>,
66}
67
68impl SqlServerTableDesc {
69    /// Creating a [`SqlServerTableDesc`] from a [`SqlServerTableRaw`] description.
70    ///
71    /// Note: Not all columns from SQL Server can be ingested into Materialize. To determine if a
72    /// column is supported see [`SqlServerColumnDesc::decode_type`].
73    pub fn new(
74        raw: SqlServerTableRaw,
75        raw_constraints: Vec<SqlServerTableConstraintRaw>,
76    ) -> Result<Self, SqlServerError> {
77        let columns: Box<[_]> = raw
78            .columns
79            .into_iter()
80            .map(SqlServerColumnDesc::new)
81            .collect();
82        let constraints = raw_constraints
83            .into_iter()
84            .map(SqlServerTableConstraint::try_from)
85            .collect::<Result<Vec<_>, _>>()?;
86        Ok(SqlServerTableDesc {
87            schema_name: raw.schema_name,
88            name: raw.name,
89            columns,
90            constraints,
91        })
92    }
93
94    /// Returns the [`SqlServerQualifiedTableName`] for this [`SqlServerTableDesc`].
95    pub fn qualified_name(&self) -> SqlServerQualifiedTableName {
96        SqlServerQualifiedTableName {
97            schema_name: Arc::clone(&self.schema_name),
98            table_name: Arc::clone(&self.name),
99        }
100    }
101
102    /// Update this [`SqlServerTableDesc`] to represent the specified columns
103    /// as text in Materialize.
104    pub fn apply_text_columns(&mut self, text_columns: &BTreeSet<&str>) {
105        for column in &mut self.columns {
106            if text_columns.contains(column.name.as_ref()) {
107                column.represent_as_text();
108            }
109        }
110    }
111
112    /// Update this [`SqlServerTableDesc`] to exclude the specified columns from being
113    /// replicated into Materialize.
114    pub fn apply_excl_columns(&mut self, excl_columns: &BTreeSet<&str>) {
115        for column in &mut self.columns {
116            if excl_columns.contains(column.name.as_ref()) {
117                column.exclude();
118            }
119        }
120    }
121
122    /// Returns a [`SqlServerRowDecoder`] which can be used to decode [`tiberius::Row`]s into
123    /// [`mz_repr::Row`]s that match the shape of the provided [`RelationDesc`].
124    pub fn decoder(&self, desc: &RelationDesc) -> Result<SqlServerRowDecoder, SqlServerError> {
125        let decoder = SqlServerRowDecoder::try_new(self, desc)?;
126        Ok(decoder)
127    }
128}
129
130impl RustType<ProtoSqlServerTableDesc> for SqlServerTableDesc {
131    fn into_proto(&self) -> ProtoSqlServerTableDesc {
132        ProtoSqlServerTableDesc {
133            name: self.name.to_string(),
134            schema_name: self.schema_name.to_string(),
135            columns: self.columns.iter().map(|c| c.into_proto()).collect(),
136            constraints: self.constraints.iter().map(|c| c.into_proto()).collect(),
137        }
138    }
139
140    fn from_proto(proto: ProtoSqlServerTableDesc) -> Result<Self, mz_proto::TryFromProtoError> {
141        let columns = proto
142            .columns
143            .into_iter()
144            .map(|c| c.into_rust())
145            .collect::<Result<_, _>>()?;
146        let constraints = proto
147            .constraints
148            .into_iter()
149            .map(|c| c.into_rust())
150            .collect::<Result<_, _>>()?;
151        Ok(SqlServerTableDesc {
152            schema_name: proto.schema_name.into(),
153            name: proto.name.into(),
154            columns,
155            constraints,
156        })
157    }
158}
159
160/// SQL Server table constraint type (e.g. PRIMARY KEY, UNIQUE, etc.)
161/// See <https://learn.microsoft.com/en-us/sql/relational-databases/system-information-schema-views/table-constraints-transact-sql?view=sql-server-ver17>
162#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
163#[cfg_attr(any(test, feature = "proptest"), derive(Arbitrary))]
164pub enum SqlServerTableConstraintType {
165    PrimaryKey,
166    Unique,
167}
168
169impl TryFrom<String> for SqlServerTableConstraintType {
170    type Error = SqlServerError;
171
172    fn try_from(value: String) -> Result<Self, Self::Error> {
173        match value.as_str() {
174            "PRIMARY KEY" => Ok(Self::PrimaryKey),
175            "UNIQUE" => Ok(Self::Unique),
176            name => Err(SqlServerError::InvalidData {
177                column_name: "constraint_type".into(),
178                error: format!("Unknown constraint type: {name}"),
179            }),
180        }
181    }
182}
183
184impl RustType<proto_sql_server_table_constraint::ConstraintType> for SqlServerTableConstraintType {
185    fn into_proto(&self) -> proto_sql_server_table_constraint::ConstraintType {
186        match self {
187            SqlServerTableConstraintType::PrimaryKey => ConstraintType::PrimaryKey(()),
188            SqlServerTableConstraintType::Unique => ConstraintType::Unique(()),
189        }
190    }
191
192    fn from_proto(
193        proto: proto_sql_server_table_constraint::ConstraintType,
194    ) -> Result<Self, mz_proto::TryFromProtoError> {
195        Ok(match proto {
196            ConstraintType::PrimaryKey(_) => SqlServerTableConstraintType::PrimaryKey,
197            ConstraintType::Unique(_) => SqlServerTableConstraintType::Unique,
198        })
199    }
200}
201
202/// SQL Server table constraint.
203#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
204#[cfg_attr(any(test, feature = "proptest"), derive(Arbitrary))]
205pub struct SqlServerTableConstraint {
206    pub constraint_name: String,
207    pub constraint_type: SqlServerTableConstraintType,
208    pub column_names: Vec<String>,
209}
210
211impl TryFrom<SqlServerTableConstraintRaw> for SqlServerTableConstraint {
212    type Error = SqlServerError;
213
214    fn try_from(value: SqlServerTableConstraintRaw) -> Result<Self, Self::Error> {
215        Ok(SqlServerTableConstraint {
216            constraint_name: value.constraint_name,
217            constraint_type: value.constraint_type.try_into()?,
218            column_names: value.columns,
219        })
220    }
221}
222
223impl RustType<ProtoSqlServerTableConstraint> for SqlServerTableConstraint {
224    fn into_proto(&self) -> ProtoSqlServerTableConstraint {
225        ProtoSqlServerTableConstraint {
226            constraint_name: self.constraint_name.clone(),
227            constraint_type: Some(self.constraint_type.into_proto()),
228            column_names: self.column_names.clone(),
229        }
230    }
231
232    fn from_proto(
233        proto: ProtoSqlServerTableConstraint,
234    ) -> Result<Self, mz_proto::TryFromProtoError> {
235        Ok(SqlServerTableConstraint {
236            constraint_name: proto.constraint_name,
237            constraint_type: proto
238                .constraint_type
239                .into_rust_if_some("ProtoSqlServerTableConstraint::constraint_type")?,
240            column_names: proto.column_names,
241        })
242    }
243}
244
245/// Partially qualified name of a table from Microsoft SQL Server.
246///
247/// TODO(sql_server3): Change this to use a &str.
248#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
249pub struct SqlServerQualifiedTableName {
250    pub schema_name: Arc<str>,
251    pub table_name: Arc<str>,
252}
253
254impl ToString for SqlServerQualifiedTableName {
255    fn to_string(&self) -> String {
256        format!(
257            "{}.{}",
258            crate::quote_identifier(&self.schema_name),
259            crate::quote_identifier(&self.table_name)
260        )
261    }
262}
263
264/// Raw metadata for a table from Microsoft SQL Server.
265///
266/// See [`SqlServerTableDesc`] for a refined description that is compatible
267/// with Materialize.
268#[derive(Debug, Clone)]
269pub struct SqlServerTableRaw {
270    /// Name of the schema the table belongs to.
271    pub schema_name: Arc<str>,
272    /// Name of the table.
273    pub name: Arc<str>,
274    /// The capture instance replicating changes.
275    pub capture_instance: Arc<SqlServerCaptureInstanceRaw>,
276    /// Columns for the table.
277    pub columns: Arc<[SqlServerColumnRaw]>,
278}
279
280/// Raw capture instance metadata.
281#[derive(Debug, Clone)]
282pub struct SqlServerCaptureInstanceRaw {
283    /// The capture instance replicating changes.
284    pub name: Arc<str>,
285    /// The creation date of the capture instance.
286    pub create_date: Arc<NaiveDateTime>,
287}
288
289/// Description of a column from a table in Microsoft SQL Server.
290#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
291#[cfg_attr(any(test, feature = "proptest"), derive(Arbitrary))]
292pub struct SqlServerColumnDesc {
293    /// Name of the column.
294    pub name: Arc<str>,
295    /// The intended data type of the this column in Materialize. `None` indicates this
296    /// column should be excluded when replicating into Materialize.
297    ///
298    /// Note: This type might differ from the `decode_type`, e.g. a user can
299    /// specify `TEXT COLUMNS` to decode columns as text.
300    pub column_type: Option<SqlColumnType>,
301    /// This field is deprecated and will be removed in a future version.  This exists only for the
302    /// purpose of migrating from old representations.
303    pub primary_key_constraint: Option<Arc<str>>,
304    /// Rust type we should parse the data from a [`tiberius::Row`] as.
305    pub decode_type: SqlServerColumnDecodeType,
306    /// Raw type of the column as we read it from upstream.
307    ///
308    /// This is useful to keep around for debugging purposes.
309    pub raw_type: Arc<str>,
310}
311
312impl SqlServerColumnDesc {
313    /// Create a [`SqlServerColumnDesc`] from a [`SqlServerColumnRaw`] description.
314    pub fn new(raw: &SqlServerColumnRaw) -> Self {
315        let (column_type, decode_type) = match parse_data_type(raw) {
316            Ok((scalar_type, decode_type)) => {
317                let column_type = scalar_type.nullable(raw.is_nullable);
318                (Some(column_type), decode_type)
319            }
320            Err(err) => {
321                tracing::warn!(
322                    ?err,
323                    ?raw,
324                    "found an unsupported data type when parsing raw data"
325                );
326                (
327                    None,
328                    SqlServerColumnDecodeType::Unsupported {
329                        context: err.reason,
330                    },
331                )
332            }
333        };
334        SqlServerColumnDesc {
335            name: Arc::clone(&raw.name),
336            primary_key_constraint: None,
337            column_type,
338            decode_type,
339            raw_type: Arc::clone(&raw.data_type),
340        }
341    }
342
343    /// Change this [`SqlServerColumnDesc`] to be represented as text in Materialize.
344    pub fn represent_as_text(&mut self) {
345        self.column_type = self
346            .column_type
347            .as_ref()
348            .map(|ct| SqlScalarType::String.nullable(ct.nullable));
349    }
350
351    /// Exclude this [`SqlServerColumnDesc`] from being replicated into Materialize.
352    pub fn exclude(&mut self) {
353        self.column_type = None;
354    }
355
356    /// Check if this [`SqlServerColumnDesc`] is excluded from being replicated into Materialize.
357    pub fn is_excluded(&self) -> bool {
358        self.column_type.is_none()
359    }
360}
361
362impl RustType<ProtoSqlServerColumnDesc> for SqlServerColumnDesc {
363    fn into_proto(&self) -> ProtoSqlServerColumnDesc {
364        ProtoSqlServerColumnDesc {
365            name: self.name.to_string(),
366            column_type: self.column_type.into_proto(),
367            primary_key_constraint: self.primary_key_constraint.as_ref().map(|v| v.to_string()),
368            decode_type: Some(self.decode_type.into_proto()),
369            raw_type: self.raw_type.to_string(),
370        }
371    }
372
373    fn from_proto(proto: ProtoSqlServerColumnDesc) -> Result<Self, mz_proto::TryFromProtoError> {
374        Ok(SqlServerColumnDesc {
375            name: proto.name.into(),
376            column_type: proto.column_type.into_rust()?,
377            primary_key_constraint: proto.primary_key_constraint.map(|v| v.into()),
378            decode_type: proto
379                .decode_type
380                .into_rust_if_some("ProtoSqlServerColumnDesc::decode_type")?,
381            raw_type: proto.raw_type.into(),
382        })
383    }
384}
385
386/// The raw datatype from SQL Server is not supported in Materialize.
387#[derive(Debug)]
388#[allow(dead_code)]
389pub struct UnsupportedDataType {
390    column_name: String,
391    column_type: String,
392    reason: String,
393}
394
395/// Parse a raw data type from SQL Server into a Materialize [`SqlScalarType`].
396///
397/// Returns the [`SqlScalarType`] that we'll map this column to and the [`SqlServerColumnDecodeType`]
398/// that we use to decode the raw value.
399fn parse_data_type(
400    raw: &SqlServerColumnRaw,
401) -> Result<(SqlScalarType, SqlServerColumnDecodeType), UnsupportedDataType> {
402    // The value of a computed column, persisted or not, will be readable by the snapshot, but will
403    // always be NULL in the CDC stream.  This can lead to issues in MZ (e.g. decoding errors,
404    // negative accumulations, etc.).
405    if raw.is_computed {
406        return Err(UnsupportedDataType {
407            column_name: raw.name.to_string(),
408            column_type: format!("{} (computed)", raw.data_type.to_lowercase()),
409            reason: "column is computed".into(),
410        });
411    }
412
413    let scalar = match raw.data_type.to_lowercase().as_str() {
414        "tinyint" => (SqlScalarType::Int16, SqlServerColumnDecodeType::U8),
415        "smallint" => (SqlScalarType::Int16, SqlServerColumnDecodeType::I16),
416        "int" => (SqlScalarType::Int32, SqlServerColumnDecodeType::I32),
417        "bigint" => (SqlScalarType::Int64, SqlServerColumnDecodeType::I64),
418        "bit" => (SqlScalarType::Bool, SqlServerColumnDecodeType::Bool),
419        "decimal" | "numeric" | "money" | "smallmoney" => {
420            // SQL Server supports a precision in the range of [1, 38] and then
421            // the scale is 0 <= scale <= precision.
422            //
423            // Materialize numerics are floating point with a fixed precision of 39.
424            //
425            // See: <https://learn.microsoft.com/en-us/sql/t-sql/data-types/decimal-and-numeric-transact-sql?view=sql-server-ver16#arguments>
426            if raw.precision > 38 || raw.scale > raw.precision {
427                tracing::warn!(
428                    "unexpected value from SQL Server, precision of {} and scale of {}",
429                    raw.precision,
430                    raw.scale,
431                );
432            }
433            if raw.precision > 39 {
434                let reason = format!(
435                    "precision of {} is greater than our maximum of 39",
436                    raw.precision
437                );
438                return Err(UnsupportedDataType {
439                    column_name: raw.name.to_string(),
440                    column_type: raw.data_type.to_string(),
441                    reason,
442                });
443            }
444
445            let raw_scale = usize::cast_from(raw.scale);
446            let max_scale =
447                NumericMaxScale::try_from(raw_scale).map_err(|_| UnsupportedDataType {
448                    column_type: raw.data_type.to_string(),
449                    column_name: raw.name.to_string(),
450                    reason: format!("scale of {} is too large", raw.scale),
451                })?;
452            let column_type = SqlScalarType::Numeric {
453                max_scale: Some(max_scale),
454            };
455
456            (column_type, SqlServerColumnDecodeType::Numeric)
457        }
458        // SQL Server has a few IEEE 754 floating point type names. The underlying type is float(n),
459        // where n is the number of bits used. SQL Server still ends up with only 2 distinct types
460        // as it treats 1 <= n <= 24 as n=24, and 25 <= n <= 53 as n=53.
461        //
462        // Additionally, `real` and `double precision` exist as synonyms of float(24) and float(53),
463        // respectively.  What doesn't appear to be documented is how these appear in `sys.types`.
464        // See <https://learn.microsoft.com/en-us/sql/t-sql/data-types/float-and-real-transact-sql?view=sql-server-ver17>
465        "real" | "float" | "double precision" => match raw.max_length {
466            // Decide the MZ type based on the number of bytes rather than the name, just in case
467            // there is inconsistency among versions.
468            4 => (SqlScalarType::Float32, SqlServerColumnDecodeType::F32),
469            8 => (SqlScalarType::Float64, SqlServerColumnDecodeType::F64),
470            _ => {
471                return Err(UnsupportedDataType {
472                    column_name: raw.name.to_string(),
473                    column_type: raw.data_type.to_string(),
474                    reason: format!("unsupported length {}", raw.max_length),
475                });
476            }
477        },
478        "char" | "nchar" | "sysname" => {
479            // There isn't a char(max) or nchar(max), so it isn't clear if this condition
480            // is possible.
481            if raw.max_length == -1 {
482                return Err(UnsupportedDataType {
483                    column_name: raw.name.to_string(),
484                    column_type: raw.data_type.to_string(),
485                    reason: "columns with unlimited size do not support CDC".to_string(),
486                });
487            }
488
489            // We represent these as "text" rather than a fixed-length
490            // `character(n)`. SQL Server sizes them in bytes, while
491            // Materialize's `Char` length is a character count. Under a
492            // multi-byte collation a single character can span several
493            // bytes, so `sys.columns.max_length` is not a usable character
494            // count. Rather than guess, we pass the value through as-is.
495            (SqlScalarType::String, SqlServerColumnDecodeType::String)
496        }
497        "varchar" | "nvarchar" => {
498            // `max text repl size` is 64KB by default.  If a user attempts to insert a value
499            // that exceeds this limit, SQL Server will return an error and the insert fails
500            // with error `7139`.  This is also true for updates that increase the field length
501            // beyond the limit.
502            //
503            // See <https://learn.microsoft.com/en-us/sql/relational-databases/errors-events/database-engine-events-and-errors-7000-to-7999?view=sql-server-ver17>
504            //
505            // If the `max text repl size` changes, it does not affect events already written to
506            // the CDC table, nor does it change the behavior of what CDC captures for updates
507            // to non-LOD columns (based on testing).
508            let max_length = if raw.max_length != -1 {
509                let length =
510                    VarCharMaxLength::try_from(i64::from(raw.max_length)).map_err(|e| {
511                        UnsupportedDataType {
512                            column_name: raw.name.to_string(),
513                            column_type: raw.data_type.to_string(),
514                            reason: e.to_string(),
515                        }
516                    })?;
517                Some(length)
518            } else {
519                None
520            };
521            let column_type = SqlScalarType::VarChar { max_length };
522            (column_type, SqlServerColumnDecodeType::String)
523        }
524        "text" | "ntext" | "image" => {
525            // SQL Server docs indicate this should always be 16. There's no
526            // issue if it's not, but it's good to track.
527            mz_ore::soft_assert_eq_no_log!(raw.max_length, 16);
528
529            // TODO(sql_server3): Support UPSERT semantics for SQL Server.
530            return Err(UnsupportedDataType {
531                column_name: raw.name.to_string(),
532                column_type: raw.data_type.to_string(),
533                reason: "columns with unlimited size do not support CDC".to_string(),
534            });
535        }
536        "xml" => {
537            // When the `max_length` is -1 SQL Server will not present us with the "before" value
538            // for updated columns.
539            //
540            // TODO(sql_server3): Support UPSERT semantics for SQL Server.
541            if raw.max_length == -1 {
542                return Err(UnsupportedDataType {
543                    column_name: raw.name.to_string(),
544                    column_type: raw.data_type.to_string(),
545                    reason: "columns with unlimited size do not support CDC".to_string(),
546                });
547            }
548            (SqlScalarType::String, SqlServerColumnDecodeType::Xml)
549        }
550        "binary" | "varbinary" => {
551            // [`SqlScalarType`] does not support tracking max_length for binary data. To ensure
552            // columns of type varbinary(max) (Large Object Data) are decoded properly, it is
553            // necessary to know that the length is `max`. `varchar` and `nvarchar` track this
554            // using [`SqlScalarType::VarChar`] max_length field.
555            if raw.max_length == -1 {
556                return Err(UnsupportedDataType {
557                    column_name: raw.name.to_string(),
558                    column_type: raw.data_type.to_string(),
559                    reason: "columns with unlimited size do not support CDC".to_string(),
560                });
561            }
562            (SqlScalarType::Bytes, SqlServerColumnDecodeType::Bytes)
563        }
564        "json" => (SqlScalarType::Jsonb, SqlServerColumnDecodeType::String),
565        "date" => (SqlScalarType::Date, SqlServerColumnDecodeType::NaiveDate),
566        // SQL Server supports a scale of (and defaults to) 7 digits (aka 100 nanoseconds)
567        // for time related types.
568        //
569        // Internally Materialize supports a scale of 9 (aka nanoseconds), but for Postgres
570        // compatibility we constraint ourselves to a scale of 6 (aka microseconds). By
571        // default we will round values we get from  SQL Server to fit in Materialize.
572        //
573        // TODO(sql_server3): Support a "strict" mode where we're fail the creation of the
574        // source if the scale is too large.
575        // TODO(sql_server3): Support specifying a precision for SqlScalarType::Time.
576        //
577        // See: <https://learn.microsoft.com/en-us/sql/t-sql/data-types/datetime2-transact-sql?view=sql-server-ver16>.
578        "time" => (SqlScalarType::Time, SqlServerColumnDecodeType::NaiveTime),
579        dt @ ("smalldatetime" | "datetime" | "datetime2" | "datetimeoffset") => {
580            if raw.scale > 7 {
581                tracing::warn!("unexpected scale '{}' from SQL Server", raw.scale);
582            }
583            if raw.scale > mz_repr::adt::timestamp::MAX_PRECISION {
584                tracing::warn!("truncating scale of '{}' for '{}'", raw.scale, dt);
585            }
586            let precision = std::cmp::min(raw.scale, mz_repr::adt::timestamp::MAX_PRECISION);
587            let precision =
588                Some(TimestampPrecision::try_from(i64::from(precision)).expect("known to fit"));
589
590            match dt {
591                "smalldatetime" | "datetime" | "datetime2" => (
592                    SqlScalarType::Timestamp { precision },
593                    SqlServerColumnDecodeType::NaiveDateTime,
594                ),
595                "datetimeoffset" => (
596                    SqlScalarType::TimestampTz { precision },
597                    SqlServerColumnDecodeType::DateTime,
598                ),
599                other => unreachable!("'{other}' checked above"),
600            }
601        }
602        "uniqueidentifier" => (SqlScalarType::Uuid, SqlServerColumnDecodeType::Uuid),
603        // TODO(sql_server3): Support reading the following types, at least as text:
604        //
605        // * geography
606        // * geometry
607        // * json (preview)
608        // * vector (preview)
609        //
610        // None of these types are implemented in `tiberius`, the crate that
611        // provides our SQL Server client, so we'll need to implement support
612        // for decoding them.
613        //
614        // See <https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-tds/355f7890-6e91-4978-ab76-2ded17ee09bc>.
615        other => {
616            return Err(UnsupportedDataType {
617                column_type: other.to_string(),
618                column_name: raw.name.to_string(),
619                reason: format!("'{other}' is unimplemented"),
620            });
621        }
622    };
623    Ok(scalar)
624}
625
626/// Raw metadata for a column from a table in Microsoft SQL Server.
627///
628/// See: <https://learn.microsoft.com/en-us/sql/relational-databases/system-catalog-views/sys-columns-transact-sql?view=sql-server-ver16>.
629#[derive(Clone, Debug)]
630pub struct SqlServerColumnRaw {
631    /// Name of this column.
632    pub name: Arc<str>,
633    /// Name of the data type.
634    pub data_type: Arc<str>,
635    /// Whether or not the column is nullable.
636    pub is_nullable: bool,
637    /// Maximum length (in bytes) of the column.
638    ///
639    /// For `varchar(max)`, `nvarchar(max)`, `varbinary(max)`, or `xml` this will be `-1`. For
640    /// `text`, `ntext`, and `image` columns this will be 16.
641    ///
642    /// See: <https://learn.microsoft.com/en-us/sql/relational-databases/system-catalog-views/sys-columns-transact-sql?view=sql-server-ver16>.
643    ///
644    /// TODO(sql_server2): Validate this value for `json` columns where were introduced
645    /// Azure SQL 2024.
646    pub max_length: i16,
647    /// Precision of the column, if numeric-based; otherwise 0.
648    pub precision: u8,
649    /// Scale of the columns, if numeric-based; otherwise 0.
650    pub scale: u8,
651    /// Whether the column is computed.
652    pub is_computed: bool,
653}
654
655/// Raw metadata for a table constraint.
656#[derive(Clone, Debug)]
657pub struct SqlServerTableConstraintRaw {
658    pub constraint_name: String,
659    pub constraint_type: String,
660    pub columns: Vec<String>,
661}
662
663/// Rust type that we should use when reading a column from SQL Server.
664#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
665#[cfg_attr(any(test, feature = "proptest"), derive(Arbitrary))]
666pub enum SqlServerColumnDecodeType {
667    Bool,
668    U8,
669    I16,
670    I32,
671    I64,
672    F32,
673    F64,
674    String,
675    Bytes,
676    /// [`uuid::Uuid`].
677    Uuid,
678    /// [`tiberius::numeric::Numeric`].
679    Numeric,
680    /// [`tiberius::xml::XmlData`].
681    Xml,
682    /// [`chrono::NaiveDate`].
683    NaiveDate,
684    /// [`chrono::NaiveTime`].
685    NaiveTime,
686    /// [`chrono::DateTime`].
687    DateTime,
688    /// [`chrono::NaiveDateTime`].
689    NaiveDateTime,
690    /// Decoding this type isn't supported.
691    Unsupported {
692        /// Any additional context as to why this type isn't supported.
693        context: String,
694    },
695}
696
697impl SqlServerColumnDecodeType {
698    /// Decode the column with `name` out of the provided `data`.
699    pub fn decode<'a>(
700        &self,
701        data: &'a tiberius::Row,
702        name: &'a str,
703        column: &'a SqlColumnType,
704        arena: &'a RowArena,
705    ) -> Result<Datum<'a>, SqlServerDecodeError> {
706        let maybe_datum = match (&column.scalar_type, self) {
707            (SqlScalarType::Bool, SqlServerColumnDecodeType::Bool) => data
708                .try_get(name)
709                .map_err(|_| SqlServerDecodeError::invalid_column(name, "bool"))?
710                .map(|val: bool| if val { Datum::True } else { Datum::False }),
711            (SqlScalarType::Int16, SqlServerColumnDecodeType::U8) => data
712                .try_get(name)
713                .map_err(|_| SqlServerDecodeError::invalid_column(name, "u8"))?
714                .map(|val: u8| Datum::Int16(i16::cast_from(val))),
715            (SqlScalarType::Int16, SqlServerColumnDecodeType::I16) => data
716                .try_get(name)
717                .map_err(|_| SqlServerDecodeError::invalid_column(name, "i16"))?
718                .map(Datum::Int16),
719            (SqlScalarType::Int32, SqlServerColumnDecodeType::I32) => data
720                .try_get(name)
721                .map_err(|_| SqlServerDecodeError::invalid_column(name, "i32"))?
722                .map(Datum::Int32),
723            (SqlScalarType::Int64, SqlServerColumnDecodeType::I64) => data
724                .try_get(name)
725                .map_err(|_| SqlServerDecodeError::invalid_column(name, "i64"))?
726                .map(Datum::Int64),
727            (SqlScalarType::Float32, SqlServerColumnDecodeType::F32) => data
728                .try_get(name)
729                .map_err(|_| SqlServerDecodeError::invalid_column(name, "f32"))?
730                .map(|val: f32| Datum::Float32(ordered_float::OrderedFloat(val))),
731            (SqlScalarType::Float64, SqlServerColumnDecodeType::F64) => data
732                .try_get(name)
733                .map_err(|_| SqlServerDecodeError::invalid_column(name, "f64"))?
734                .map(|val: f64| Datum::Float64(ordered_float::OrderedFloat(val))),
735            (SqlScalarType::String, SqlServerColumnDecodeType::String) => data
736                .try_get(name)
737                .map_err(|_| SqlServerDecodeError::invalid_column(name, "string"))?
738                .map(Datum::String),
739            // `char`, `nchar`, and `sysname` columns now map to `String`, so
740            // new sources never produce this type. It remains to decode sources
741            // created before that change, whose persisted desc still carries
742            // `Char { length }`.
743            (SqlScalarType::Char { length }, SqlServerColumnDecodeType::String) => data
744                .try_get(name)
745                .map_err(|_| SqlServerDecodeError::invalid_column(name, "char"))?
746                .map(|val: &str| match length {
747                    // SQL Server sizes `char(n)` in bytes, and `length` (from
748                    // `sys.columns.max_length`) is that byte count, while
749                    // Materialize's `Char` length is a character count. SQL
750                    // Server blank-pads the value to fill the byte length, so a
751                    // multi-byte collation yields fewer characters than bytes.
752                    // The character count can therefore be at most `length`,
753                    // never more, so we only reject when it exceeds `length`.
754                    //
755                    // This encodes at a different length in Materialize vs.
756                    // the upstream, which is why it is deprecated. An
757                    // upstream CHAR(10) column with the string "café" would
758                    // have 5 trailing spaces. In Materialize, this would
759                    // have 6 trailing spaces.
760                    Some(max) => {
761                        let found_chars = val.chars().count();
762                        let max_chars = usize::cast_from(max.into_u32());
763                        if found_chars > max_chars {
764                            Err(SqlServerDecodeError::invalid_char(
765                                name,
766                                max_chars,
767                                found_chars,
768                            ))
769                        } else {
770                            Ok(Datum::String(val))
771                        }
772                    }
773                    None => Ok(Datum::String(val)),
774                })
775                .transpose()?,
776            (SqlScalarType::VarChar { max_length }, SqlServerColumnDecodeType::String) => data
777                .try_get(name)
778                .map_err(|_| SqlServerDecodeError::invalid_column(name, "varchar"))?
779                .map(|val: &str| match max_length {
780                    Some(max) => {
781                        let found_chars = val.chars().count();
782                        let max_chars = usize::cast_from(max.into_u32());
783                        if found_chars > max_chars {
784                            Err(SqlServerDecodeError::invalid_varchar(
785                                name,
786                                max_chars,
787                                found_chars,
788                            ))
789                        } else {
790                            Ok(Datum::String(val))
791                        }
792                    }
793                    None => Ok(Datum::String(val)),
794                })
795                .transpose()?,
796            (SqlScalarType::Bytes, SqlServerColumnDecodeType::Bytes) => data
797                .try_get(name)
798                .map_err(|_| SqlServerDecodeError::invalid_column(name, "bytes"))?
799                .map(Datum::Bytes),
800            (SqlScalarType::Uuid, SqlServerColumnDecodeType::Uuid) => data
801                .try_get(name)
802                .map_err(|_| SqlServerDecodeError::invalid_column(name, "uuid"))?
803                .map(Datum::Uuid),
804            (SqlScalarType::Numeric { .. }, SqlServerColumnDecodeType::Numeric) => data
805                .try_get(name)
806                .map_err(|_| SqlServerDecodeError::invalid_column(name, "numeric"))?
807                .map(|val: tiberius::numeric::Numeric| {
808                    let numeric = tiberius_numeric_to_mz_numeric(val);
809                    Datum::Numeric(OrderedDecimal(numeric))
810                }),
811            (SqlScalarType::String, SqlServerColumnDecodeType::Xml) => data
812                .try_get(name)
813                .map_err(|_| SqlServerDecodeError::invalid_column(name, "xml"))?
814                .map(|val: &tiberius::xml::XmlData| Datum::String(val.as_ref())),
815            (SqlScalarType::Date, SqlServerColumnDecodeType::NaiveDate) => data
816                .try_get(name)
817                .map_err(|_| SqlServerDecodeError::invalid_column(name, "date"))?
818                .map(|val: chrono::NaiveDate| {
819                    let date = val
820                        .try_into()
821                        .map_err(|e| SqlServerDecodeError::invalid_date(name, e))?;
822                    Ok::<_, SqlServerDecodeError>(Datum::Date(date))
823                })
824                .transpose()?,
825            (SqlScalarType::Time, SqlServerColumnDecodeType::NaiveTime) => data
826                .try_get(name)
827                .map_err(|_| SqlServerDecodeError::invalid_column(name, "time"))?
828                .map(|val: chrono::NaiveTime| {
829                    // Postgres' maximum precision is 6 (aka microseconds).
830                    //
831                    // While the Postgres spec supports specifying a precision
832                    // Materialize does not.
833                    let rounded = val.round_subsecs(6);
834                    // Overflowed.
835                    let val = if rounded < val {
836                        val.trunc_subsecs(6)
837                    } else {
838                        val
839                    };
840                    Datum::Time(val)
841                }),
842            (SqlScalarType::Timestamp { precision }, SqlServerColumnDecodeType::NaiveDateTime) => {
843                data.try_get(name)
844                    .map_err(|_| SqlServerDecodeError::invalid_column(name, "timestamp"))?
845                    .map(|val: chrono::NaiveDateTime| {
846                        let ts: CheckedTimestamp<chrono::NaiveDateTime> = val
847                            .try_into()
848                            .map_err(|e| SqlServerDecodeError::invalid_timestamp(name, e))?;
849                        let rounded = ts
850                            .round_to_precision(*precision)
851                            .map_err(|e| SqlServerDecodeError::invalid_timestamp(name, e))?;
852                        Ok::<_, SqlServerDecodeError>(Datum::Timestamp(rounded))
853                    })
854                    .transpose()?
855            }
856            (SqlScalarType::TimestampTz { precision }, SqlServerColumnDecodeType::DateTime) => data
857                .try_get(name)
858                .map_err(|_| SqlServerDecodeError::invalid_column(name, "timestamptz"))?
859                .map(|val: chrono::DateTime<chrono::Utc>| {
860                    let ts: CheckedTimestamp<chrono::DateTime<chrono::Utc>> = val
861                        .try_into()
862                        .map_err(|e| SqlServerDecodeError::invalid_timestamp(name, e))?;
863                    let rounded = ts
864                        .round_to_precision(*precision)
865                        .map_err(|e| SqlServerDecodeError::invalid_timestamp(name, e))?;
866                    Ok::<_, SqlServerDecodeError>(Datum::TimestampTz(rounded))
867                })
868                .transpose()?,
869            // We support mapping any type to a string.
870            (SqlScalarType::String, SqlServerColumnDecodeType::Bool) => data
871                .try_get(name)
872                .map_err(|_| SqlServerDecodeError::invalid_column(name, "bool-text"))?
873                .map(|val: bool| {
874                    if val {
875                        Datum::String("true")
876                    } else {
877                        Datum::String("false")
878                    }
879                }),
880            (SqlScalarType::String, SqlServerColumnDecodeType::U8) => data
881                .try_get(name)
882                .map_err(|_| SqlServerDecodeError::invalid_column(name, "u8-text"))?
883                .map(|val: u8| {
884                    arena.make_datum(|packer| packer.push(Datum::String(&val.to_string())))
885                }),
886            (SqlScalarType::String, SqlServerColumnDecodeType::I16) => data
887                .try_get(name)
888                .map_err(|_| SqlServerDecodeError::invalid_column(name, "i16-text"))?
889                .map(|val: i16| {
890                    arena.make_datum(|packer| packer.push(Datum::String(&val.to_string())))
891                }),
892            (SqlScalarType::String, SqlServerColumnDecodeType::I32) => data
893                .try_get(name)
894                .map_err(|_| SqlServerDecodeError::invalid_column(name, "i32-text"))?
895                .map(|val: i32| {
896                    arena.make_datum(|packer| packer.push(Datum::String(&val.to_string())))
897                }),
898            (SqlScalarType::String, SqlServerColumnDecodeType::I64) => data
899                .try_get(name)
900                .map_err(|_| SqlServerDecodeError::invalid_column(name, "i64-text"))?
901                .map(|val: i64| {
902                    arena.make_datum(|packer| packer.push(Datum::String(&val.to_string())))
903                }),
904            (SqlScalarType::String, SqlServerColumnDecodeType::F32) => data
905                .try_get(name)
906                .map_err(|_| SqlServerDecodeError::invalid_column(name, "f32-text"))?
907                .map(|val: f32| {
908                    arena.make_datum(|packer| packer.push(Datum::String(&val.to_string())))
909                }),
910            (SqlScalarType::String, SqlServerColumnDecodeType::F64) => data
911                .try_get(name)
912                .map_err(|_| SqlServerDecodeError::invalid_column(name, "f64-text"))?
913                .map(|val: f64| {
914                    arena.make_datum(|packer| packer.push(Datum::String(&val.to_string())))
915                }),
916            (SqlScalarType::String, SqlServerColumnDecodeType::Uuid) => data
917                .try_get(name)
918                .map_err(|_| SqlServerDecodeError::invalid_column(name, "uuid-text"))?
919                .map(|val: uuid::Uuid| {
920                    arena.make_datum(|packer| packer.push(Datum::String(&val.to_string())))
921                }),
922            (SqlScalarType::String, SqlServerColumnDecodeType::Bytes) => data
923                .try_get(name)
924                .map_err(|_| SqlServerDecodeError::invalid_column(name, "bytes-text"))?
925                .map(|val: &[u8]| {
926                    let encoded = base64::engine::general_purpose::STANDARD.encode(val);
927                    arena.make_datum(|packer| packer.push(Datum::String(&encoded)))
928                }),
929            (SqlScalarType::String, SqlServerColumnDecodeType::Numeric) => data
930                .try_get(name)
931                .map_err(|_| SqlServerDecodeError::invalid_column(name, "numeric-text"))?
932                .map(|val: tiberius::numeric::Numeric| {
933                    arena.make_datum(|packer| packer.push(Datum::String(&val.to_string())))
934                }),
935            (SqlScalarType::String, SqlServerColumnDecodeType::NaiveDate) => data
936                .try_get(name)
937                .map_err(|_| SqlServerDecodeError::invalid_column(name, "naivedate-text"))?
938                .map(|val: chrono::NaiveDate| {
939                    arena.make_datum(|packer| packer.push(Datum::String(&val.to_string())))
940                }),
941            (SqlScalarType::String, SqlServerColumnDecodeType::NaiveTime) => data
942                .try_get(name)
943                .map_err(|_| SqlServerDecodeError::invalid_column(name, "naivetime-text"))?
944                .map(|val: chrono::NaiveTime| {
945                    arena.make_datum(|packer| packer.push(Datum::String(&val.to_string())))
946                }),
947            (SqlScalarType::String, SqlServerColumnDecodeType::DateTime) => data
948                .try_get(name)
949                .map_err(|_| SqlServerDecodeError::invalid_column(name, "datetime-text"))?
950                .map(|val: chrono::DateTime<chrono::Utc>| {
951                    arena.make_datum(|packer| packer.push(Datum::String(&val.to_string())))
952                }),
953            (SqlScalarType::String, SqlServerColumnDecodeType::NaiveDateTime) => data
954                .try_get(name)
955                .map_err(|_| SqlServerDecodeError::invalid_column(name, "naivedatetime-text"))?
956                .map(|val: chrono::NaiveDateTime| {
957                    arena.make_datum(|packer| packer.push(Datum::String(&val.to_string())))
958                }),
959            (column_type, decode_type) => {
960                return Err(SqlServerDecodeError::Unsupported {
961                    sql_server_type: decode_type.clone(),
962                    mz_type: column_type.clone(),
963                });
964            }
965        };
966
967        match (maybe_datum, column.nullable) {
968            (Some(datum), _) => Ok(datum),
969            (None, true) => Ok(Datum::Null),
970            (None, false) => Err(SqlServerDecodeError::InvalidData {
971                column_name: name.to_string(),
972                // Note: This error string is durably recorded in Persist, do not change.
973                error: "found Null in non-nullable column".to_string(),
974            }),
975        }
976    }
977}
978
979impl RustType<proto_sql_server_column_desc::DecodeType> for SqlServerColumnDecodeType {
980    fn into_proto(&self) -> proto_sql_server_column_desc::DecodeType {
981        match self {
982            SqlServerColumnDecodeType::Bool => proto_sql_server_column_desc::DecodeType::Bool(()),
983            SqlServerColumnDecodeType::U8 => proto_sql_server_column_desc::DecodeType::U8(()),
984            SqlServerColumnDecodeType::I16 => proto_sql_server_column_desc::DecodeType::I16(()),
985            SqlServerColumnDecodeType::I32 => proto_sql_server_column_desc::DecodeType::I32(()),
986            SqlServerColumnDecodeType::I64 => proto_sql_server_column_desc::DecodeType::I64(()),
987            SqlServerColumnDecodeType::F32 => proto_sql_server_column_desc::DecodeType::F32(()),
988            SqlServerColumnDecodeType::F64 => proto_sql_server_column_desc::DecodeType::F64(()),
989            SqlServerColumnDecodeType::String => {
990                proto_sql_server_column_desc::DecodeType::String(())
991            }
992            SqlServerColumnDecodeType::Bytes => proto_sql_server_column_desc::DecodeType::Bytes(()),
993            SqlServerColumnDecodeType::Uuid => proto_sql_server_column_desc::DecodeType::Uuid(()),
994            SqlServerColumnDecodeType::Numeric => {
995                proto_sql_server_column_desc::DecodeType::Numeric(())
996            }
997            SqlServerColumnDecodeType::Xml => proto_sql_server_column_desc::DecodeType::Xml(()),
998            SqlServerColumnDecodeType::NaiveDate => {
999                proto_sql_server_column_desc::DecodeType::NaiveDate(())
1000            }
1001            SqlServerColumnDecodeType::NaiveTime => {
1002                proto_sql_server_column_desc::DecodeType::NaiveTime(())
1003            }
1004            SqlServerColumnDecodeType::DateTime => {
1005                proto_sql_server_column_desc::DecodeType::DateTime(())
1006            }
1007            SqlServerColumnDecodeType::NaiveDateTime => {
1008                proto_sql_server_column_desc::DecodeType::NaiveDateTime(())
1009            }
1010            SqlServerColumnDecodeType::Unsupported { context } => {
1011                proto_sql_server_column_desc::DecodeType::Unsupported(context.clone())
1012            }
1013        }
1014    }
1015
1016    fn from_proto(
1017        proto: proto_sql_server_column_desc::DecodeType,
1018    ) -> Result<Self, mz_proto::TryFromProtoError> {
1019        let val = match proto {
1020            proto_sql_server_column_desc::DecodeType::Bool(()) => SqlServerColumnDecodeType::Bool,
1021            proto_sql_server_column_desc::DecodeType::U8(()) => SqlServerColumnDecodeType::U8,
1022            proto_sql_server_column_desc::DecodeType::I16(()) => SqlServerColumnDecodeType::I16,
1023            proto_sql_server_column_desc::DecodeType::I32(()) => SqlServerColumnDecodeType::I32,
1024            proto_sql_server_column_desc::DecodeType::I64(()) => SqlServerColumnDecodeType::I64,
1025            proto_sql_server_column_desc::DecodeType::F32(()) => SqlServerColumnDecodeType::F32,
1026            proto_sql_server_column_desc::DecodeType::F64(()) => SqlServerColumnDecodeType::F64,
1027            proto_sql_server_column_desc::DecodeType::String(()) => {
1028                SqlServerColumnDecodeType::String
1029            }
1030            proto_sql_server_column_desc::DecodeType::Bytes(()) => SqlServerColumnDecodeType::Bytes,
1031            proto_sql_server_column_desc::DecodeType::Uuid(()) => SqlServerColumnDecodeType::Uuid,
1032            proto_sql_server_column_desc::DecodeType::Numeric(()) => {
1033                SqlServerColumnDecodeType::Numeric
1034            }
1035            proto_sql_server_column_desc::DecodeType::Xml(()) => SqlServerColumnDecodeType::Xml,
1036            proto_sql_server_column_desc::DecodeType::NaiveDate(()) => {
1037                SqlServerColumnDecodeType::NaiveDate
1038            }
1039            proto_sql_server_column_desc::DecodeType::NaiveTime(()) => {
1040                SqlServerColumnDecodeType::NaiveTime
1041            }
1042            proto_sql_server_column_desc::DecodeType::DateTime(()) => {
1043                SqlServerColumnDecodeType::DateTime
1044            }
1045            proto_sql_server_column_desc::DecodeType::NaiveDateTime(()) => {
1046                SqlServerColumnDecodeType::NaiveDateTime
1047            }
1048            proto_sql_server_column_desc::DecodeType::Unsupported(context) => {
1049                SqlServerColumnDecodeType::Unsupported { context }
1050            }
1051        };
1052        Ok(val)
1053    }
1054}
1055
1056/// Numerics in SQL Server have a maximum precision of 38 digits, where [`Numeric`]s in
1057/// Materialize have a maximum precision of 39 digits, so this conversion is infallible.
1058fn tiberius_numeric_to_mz_numeric(val: tiberius::numeric::Numeric) -> Numeric {
1059    let mut numeric = mz_repr::adt::numeric::cx_datum().from_i128(val.value());
1060    // Use scaleb to adjust the exponent directly, avoiding precision loss from division
1061    // scaleb(x, -n) computes x * 10^(-n)
1062    mz_repr::adt::numeric::cx_datum().scaleb(&mut numeric, &Numeric::from(-i32::from(val.scale())));
1063    numeric
1064}
1065
1066/// The update mask of a CDC event row, returned by `cdc.fn_cdc_get_all_changes_<capture_instance>`
1067/// as `__$update_mask`.
1068///
1069/// See <https://learn.microsoft.com/en-us/sql/relational-databases/system-functions/cdc-fn-cdc-get-all-changes-capture-instance-transact-sql?view=sql-server-ver17>
1070#[derive(Debug)]
1071pub struct UpdateMask {
1072    mask: Vec<u8>,
1073}
1074
1075impl TryFrom<&tiberius::Row> for UpdateMask {
1076    type Error = SqlServerDecodeError;
1077
1078    fn try_from(row: &tiberius::Row) -> Result<Self, Self::Error> {
1079        static UPDATE_MASK: &str = "__$update_mask";
1080
1081        let mask: Vec<u8> = row
1082            .try_get::<&[u8], _>(UPDATE_MASK)
1083            .inspect_err(|e| tracing::warn!("Failed extracting update mask: {e:?}"))
1084            .map_err(|_| SqlServerDecodeError::InvalidColumn {
1085                column_name: UPDATE_MASK.to_string(),
1086                as_type: "bytes",
1087            })?
1088            .ok_or_else(|| SqlServerDecodeError::InvalidData {
1089                column_name: UPDATE_MASK.to_string(),
1090                error: "column cannot be null".to_string(),
1091            })?
1092            .into();
1093        Ok(UpdateMask { mask })
1094    }
1095}
1096
1097impl UpdateMask {
1098    /// Returns true if the data column was updated, false otherwise.
1099    ///
1100    /// This function panics if `col_index` exceeds the mask.
1101    ///
1102    /// The [`tiberius::Row`] returned by `cdc.fn_cdc_get_all_changes_<capture_instance>` contains
1103    /// 4 metadata columns used by CDC:
1104    /// - `__$start_lsn`
1105    /// - `__$seqval`
1106    /// - `__$operation`
1107    /// - `__$update_mask`
1108    ///
1109    /// This function will always return false for the first 4 columns.
1110    pub fn data_col_updated(&self, col_index: usize) -> bool {
1111        const CDC_METADATA_COL_COUNT: usize = 4;
1112
1113        if col_index < CDC_METADATA_COL_COUNT {
1114            return false;
1115        }
1116        let adj_col_index = col_index - CDC_METADATA_COL_COUNT;
1117        let byte_offset = adj_col_index / usize::cast_from(u8::BITS);
1118        assert!(
1119            byte_offset < self.mask.len(),
1120            "byte_offset = {byte_offset} mask_len = {}",
1121            self.mask.len()
1122        );
1123        let bit_offset = adj_col_index % usize::cast_from(u8::BITS);
1124        (self.mask[self.mask.len() - byte_offset - 1] >> bit_offset) & 1 == 1
1125    }
1126}
1127
1128/// A decoder from [`tiberius::Row`] to [`mz_repr::Row`].
1129///
1130/// The goal of this type is to perform any expensive "downcasts" so in the hot
1131/// path of decoding rows we do the minimal amount of work.
1132#[derive(Debug)]
1133pub struct SqlServerRowDecoder {
1134    decoders: Vec<(Arc<str>, SqlColumnType, SqlServerColumnDecodeType)>,
1135}
1136
1137impl SqlServerRowDecoder {
1138    /// Try to create a [`SqlServerRowDecoder`] that will decode [`tiberius::Row`]s that match
1139    /// the shape of the provided [`SqlServerTableDesc`], to [`mz_repr::Row`]s that match the
1140    /// shape of the provided [`RelationDesc`].
1141    pub fn try_new(
1142        table: &SqlServerTableDesc,
1143        desc: &RelationDesc,
1144    ) -> Result<Self, SqlServerError> {
1145        let decoders = desc
1146            .iter()
1147            .map(|(col_name, col_type)| {
1148                let sql_server_col = table
1149                    .columns
1150                    .iter()
1151                    .find(|col| col.name.as_ref() == col_name.as_str())
1152                    .ok_or_else(|| {
1153                        // TODO(sql_server2): Structured Error.
1154                        anyhow::anyhow!("no SQL Server column with name {col_name} found")
1155                    })?;
1156                let Some(sql_server_col_typ) = sql_server_col.column_type.as_ref() else {
1157                    return Err(SqlServerError::ProgrammingError(format!(
1158                        "programming error, {col_name} should have been exluded",
1159                    )));
1160                };
1161
1162                // This shouldn't be true, but be defensive.
1163                //
1164                // TODO(sql_server2): Maybe allow the Materialize column type to be
1165                // more nullable than our decoding type?
1166                //
1167                // Sad. Our timestamp types don't roundtrip their precision through
1168                // parsing so we ignore the mismatch here.
1169                let matches = match (&sql_server_col_typ.scalar_type, &col_type.scalar_type) {
1170                    (SqlScalarType::Timestamp { .. }, SqlScalarType::Timestamp { .. })
1171                    | (SqlScalarType::TimestampTz { .. }, SqlScalarType::TimestampTz { .. }) => {
1172                        // Types match so check nullability.
1173                        sql_server_col_typ.nullable == col_type.nullable
1174                    }
1175                    (_, _) => sql_server_col_typ == col_type,
1176                };
1177                if !matches {
1178                    return Err(SqlServerError::ProgrammingError(format!(
1179                        "programming error, {col_name} has mismatched type {:?} vs {:?}",
1180                        sql_server_col.column_type, col_type
1181                    )));
1182                }
1183
1184                let name = Arc::clone(&sql_server_col.name);
1185                let decoder = sql_server_col.decode_type.clone();
1186                // Note: We specifically use the `SqlColumnType` from the SqlServerTableDesc
1187                // because it retains precision.
1188                //
1189                // See: <https://github.com/MaterializeInc/database-issues/issues/3179>.
1190                let col_typ = sql_server_col_typ.clone();
1191
1192                Ok::<_, SqlServerError>((name, col_typ, decoder))
1193            })
1194            .collect::<Result<_, _>>()?;
1195
1196        Ok(SqlServerRowDecoder { decoders })
1197    }
1198
1199    /// Decode data from the provided [`tiberius::Row`] into the provided [`Row`].
1200    ///
1201    /// For updates, the new row data is provided in the event the data contains Large Object Data
1202    /// (e.g. varchar(max)). [`SqlServerRowDecoder::decode()`] will decode the [`UpdateMask`] from
1203    /// the new row and retrieve LOD values from the new row for any LOD column that was not updated
1204    /// in the old row.
1205    pub fn decode(
1206        &self,
1207        data: &tiberius::Row,
1208        row: &mut Row,
1209        arena: &RowArena,
1210        new_data: Option<&tiberius::Row>,
1211    ) -> Result<(), SqlServerDecodeError> {
1212        let mut packer = row.packer();
1213
1214        for (col_name, col_type, decoder) in &self.decoders {
1215            let datum = decoder.decode(data, col_name, col_type, arena)?;
1216
1217            let datum = if let Some(new_data) = new_data
1218                && matches!(
1219                    col_type.scalar_type,
1220                    SqlScalarType::VarChar { max_length: None }
1221                )
1222                && matches!(datum, Datum::Null)
1223            {
1224                let update_mask = UpdateMask::try_from(new_data)?;
1225                let col_index = new_data
1226                    .columns()
1227                    .iter()
1228                    .position(|c| c.name() == col_name.as_ref())
1229                    .expect("column exists");
1230                // The only time it is valid to pull the LOD column value from the new row
1231                // is if the LOD column was *not* updated. The mask check is necessary to
1232                // differentiate between updating a non-LOD column (the LOD column in old row
1233                // is NULL) and updating a LOD column where the old value is NULL.
1234                if !update_mask.data_col_updated(col_index) {
1235                    decoder.decode(new_data, col_name, col_type, arena)?
1236                } else {
1237                    datum
1238                }
1239            } else {
1240                datum
1241            };
1242
1243            packer.push(datum);
1244        }
1245        Ok(())
1246    }
1247
1248    pub fn included_column_names(&self) -> Vec<Arc<str>> {
1249        self.decoders
1250            .iter()
1251            .map(|decoder| Arc::clone(&decoder.0))
1252            .collect()
1253    }
1254}
1255
1256#[cfg(test)]
1257mod tests {
1258    use std::collections::BTreeSet;
1259    use std::sync::Arc;
1260
1261    use chrono::NaiveDateTime;
1262    use itertools::Itertools;
1263    use mz_ore::assert_contains;
1264    use mz_ore::collections::CollectionExt;
1265    use mz_repr::adt::char::CharLength;
1266    use mz_repr::adt::numeric::NumericMaxScale;
1267    use mz_repr::adt::varchar::VarCharMaxLength;
1268    use mz_repr::{Datum, RelationDesc, Row, RowArena, SqlScalarType};
1269    use tiberius::RowTestExt;
1270
1271    use crate::desc::{
1272        SqlServerCaptureInstanceRaw, SqlServerColumnDecodeType, SqlServerColumnDesc,
1273        SqlServerTableDesc, SqlServerTableRaw, tiberius_numeric_to_mz_numeric,
1274    };
1275
1276    use super::SqlServerColumnRaw;
1277
1278    impl SqlServerColumnRaw {
1279        /// Create a new [`SqlServerColumnRaw`]. The specified `data_type` is
1280        /// _not_ checked for validity.
1281        fn new(name: &str, data_type: &str) -> Self {
1282            SqlServerColumnRaw {
1283                name: name.into(),
1284                data_type: data_type.into(),
1285                is_nullable: false,
1286                max_length: 0,
1287                precision: 0,
1288                scale: 0,
1289                is_computed: false,
1290            }
1291        }
1292
1293        fn nullable(mut self, nullable: bool) -> Self {
1294            self.is_nullable = nullable;
1295            self
1296        }
1297
1298        fn max_length(mut self, max_length: i16) -> Self {
1299            self.max_length = max_length;
1300            self
1301        }
1302
1303        fn precision(mut self, precision: u8) -> Self {
1304            self.precision = precision;
1305            self
1306        }
1307
1308        fn scale(mut self, scale: u8) -> Self {
1309            self.scale = scale;
1310            self
1311        }
1312    }
1313
1314    #[mz_ore::test]
1315    fn smoketest_column_raw() {
1316        let raw = SqlServerColumnRaw::new("foo", "bit");
1317        let col = SqlServerColumnDesc::new(&raw);
1318
1319        assert_eq!(&*col.name, "foo");
1320        assert_eq!(col.column_type, Some(SqlScalarType::Bool.nullable(false)));
1321        assert_eq!(col.decode_type, SqlServerColumnDecodeType::Bool);
1322
1323        let raw = SqlServerColumnRaw::new("foo", "decimal")
1324            .precision(20)
1325            .scale(10);
1326        let col = SqlServerColumnDesc::new(&raw);
1327
1328        let col_type = SqlScalarType::Numeric {
1329            max_scale: Some(NumericMaxScale::try_from(10i64).expect("known valid")),
1330        }
1331        .nullable(false);
1332        assert_eq!(col.column_type, Some(col_type));
1333        assert_eq!(col.decode_type, SqlServerColumnDecodeType::Numeric);
1334    }
1335
1336    #[mz_ore::test]
1337    fn smoketest_column_raw_invalid() {
1338        let raw = SqlServerColumnRaw::new("foo", "bad_data_type");
1339        let desc = SqlServerColumnDesc::new(&raw);
1340        let SqlServerColumnDecodeType::Unsupported { context } = desc.decode_type else {
1341            panic!("unexpected decode type {desc:?}");
1342        };
1343        assert_contains!(context, "'bad_data_type' is unimplemented");
1344
1345        let raw = SqlServerColumnRaw::new("foo", "decimal")
1346            .precision(100)
1347            .scale(10);
1348        let desc = SqlServerColumnDesc::new(&raw);
1349        assert!(matches!(
1350            desc.decode_type,
1351            SqlServerColumnDecodeType::Unsupported { .. }
1352        ));
1353
1354        let raw = SqlServerColumnRaw::new("foo", "varbinary").max_length(-1);
1355        let desc = SqlServerColumnDesc::new(&raw);
1356        let SqlServerColumnDecodeType::Unsupported { context } = desc.decode_type else {
1357            panic!("unexpected decode type {desc:?}");
1358        };
1359        assert_contains!(context, "columns with unlimited size do not support CDC");
1360    }
1361
1362    #[mz_ore::test]
1363    fn smoketest_decoder() {
1364        let sql_server_columns = [
1365            SqlServerColumnRaw::new("a", "varchar").max_length(16),
1366            SqlServerColumnRaw::new("b", "int").nullable(true),
1367            SqlServerColumnRaw::new("c", "bit"),
1368        ];
1369        let sql_server_desc = SqlServerTableRaw {
1370            schema_name: "my_schema".into(),
1371            name: "my_table".into(),
1372            capture_instance: Arc::new(SqlServerCaptureInstanceRaw {
1373                name: "my_table_CT".into(),
1374                create_date: NaiveDateTime::parse_from_str(
1375                    "2024-01-01 00:00:00",
1376                    "%Y-%m-%d %H:%M:%S",
1377                )
1378                .unwrap()
1379                .into(),
1380            }),
1381            columns: sql_server_columns.into(),
1382        };
1383        let sql_server_desc = SqlServerTableDesc::new(sql_server_desc, vec![]).unwrap();
1384
1385        let max_length = Some(VarCharMaxLength::try_from(16).unwrap());
1386        let relation_desc = RelationDesc::builder()
1387            .with_column("a", SqlScalarType::VarChar { max_length }.nullable(false))
1388            // Note: In the upstream table 'c' is ordered after 'b'.
1389            .with_column("c", SqlScalarType::Bool.nullable(false))
1390            .with_column("b", SqlScalarType::Int32.nullable(true))
1391            .finish();
1392
1393        // This decoder should shape the SQL Server Rows into Rows compatible with the RelationDesc.
1394        let decoder = sql_server_desc
1395            .decoder(&relation_desc)
1396            .expect("known valid");
1397
1398        let sql_server_columns = [
1399            tiberius::Column::new("a".to_string(), tiberius::ColumnType::BigVarChar),
1400            tiberius::Column::new("b".to_string(), tiberius::ColumnType::Int4),
1401            tiberius::Column::new("c".to_string(), tiberius::ColumnType::Bit),
1402        ];
1403
1404        let data_a = [
1405            tiberius::ColumnData::String(Some("hello world".into())),
1406            tiberius::ColumnData::I32(Some(42)),
1407            tiberius::ColumnData::Bit(Some(true)),
1408        ];
1409        let sql_server_row_a = tiberius::Row::build(
1410            sql_server_columns
1411                .iter()
1412                .cloned()
1413                .zip_eq(data_a.into_iter()),
1414        );
1415
1416        let data_b = [
1417            tiberius::ColumnData::String(Some("foo bar".into())),
1418            tiberius::ColumnData::I32(None),
1419            tiberius::ColumnData::Bit(Some(false)),
1420        ];
1421        let sql_server_row_b =
1422            tiberius::Row::build(sql_server_columns.into_iter().zip_eq(data_b.into_iter()));
1423
1424        let mut rnd_row = Row::default();
1425        let arena = RowArena::default();
1426
1427        decoder
1428            .decode(&sql_server_row_a, &mut rnd_row, &arena, None)
1429            .unwrap();
1430        assert_eq!(
1431            &rnd_row,
1432            &Row::pack_slice(&[Datum::String("hello world"), Datum::True, Datum::Int32(42)])
1433        );
1434
1435        decoder
1436            .decode(&sql_server_row_b, &mut rnd_row, &arena, None)
1437            .unwrap();
1438        assert_eq!(
1439            &rnd_row,
1440            &Row::pack_slice(&[Datum::String("foo bar"), Datum::False, Datum::Null])
1441        );
1442    }
1443
1444    #[mz_ore::test]
1445    fn decode_legacy_char_column() {
1446        // Sources created when `char` columns mapped to `Char { length }`
1447        // carry that type in their persisted desc. `length` is the upstream
1448        // byte count, so under a multi-byte collation the blank-padded value
1449        // holds fewer characters than `length` and must still decode.
1450        let length = Some(CharLength::try_from(10i64).expect("known valid"));
1451        let column_type = SqlScalarType::Char { length }.nullable(false);
1452        let columns = [SqlServerColumnDesc {
1453            name: "a".into(),
1454            column_type: Some(column_type.clone()),
1455            primary_key_constraint: None,
1456            decode_type: SqlServerColumnDecodeType::String,
1457            raw_type: "char".into(),
1458        }];
1459        let table_desc = SqlServerTableDesc {
1460            schema_name: "my_schema".into(),
1461            name: "my_table".into(),
1462            columns: columns.into(),
1463            constraints: vec![],
1464        };
1465        let relation_desc = RelationDesc::builder()
1466            .with_column("a", column_type)
1467            .finish();
1468        let decoder = table_desc.decoder(&relation_desc).expect("known valid");
1469
1470        let char_row = |val: &'static str| {
1471            tiberius::Row::build([(
1472                tiberius::Column::new("a".to_string(), tiberius::ColumnType::BigChar),
1473                tiberius::ColumnData::String(Some(val.into())),
1474            )])
1475        };
1476        let mut mz_row = Row::default();
1477        let arena = RowArena::default();
1478
1479        // 'café' in a char(10) column under a UTF-8 collation: 10 bytes but
1480        // only 9 characters.
1481        decoder
1482            .decode(&char_row("café     "), &mut mz_row, &arena, None)
1483            .unwrap();
1484        assert_eq!(&mz_row, &Row::pack_slice(&[Datum::String("café     ")]));
1485
1486        // A single-byte collation pads to exactly `length` characters.
1487        decoder
1488            .decode(&char_row("0123456789"), &mut mz_row, &arena, None)
1489            .unwrap();
1490        assert_eq!(&mz_row, &Row::pack_slice(&[Datum::String("0123456789")]));
1491
1492        // More characters than the upstream byte count is impossible, reject.
1493        let err = decoder
1494            .decode(&char_row("0123456789!"), &mut mz_row, &arena, None)
1495            .unwrap_err();
1496        assert_contains!(err.to_string(), "expected 10 chars found 11");
1497    }
1498
1499    #[mz_ore::test]
1500    fn smoketest_decode_to_string() {
1501        #[track_caller]
1502        fn testcase(
1503            data_type: &'static str,
1504            col_type: tiberius::ColumnType,
1505            col_data: tiberius::ColumnData<'static>,
1506        ) {
1507            let columns = [SqlServerColumnRaw::new("a", data_type)];
1508            let sql_server_desc = SqlServerTableRaw {
1509                schema_name: "my_schema".into(),
1510                name: "my_table".into(),
1511                capture_instance: Arc::new(SqlServerCaptureInstanceRaw {
1512                    name: "my_table_CT".into(),
1513                    create_date: NaiveDateTime::parse_from_str(
1514                        "2024-01-01 00:00:00",
1515                        "%Y-%m-%d %H:%M:%S",
1516                    )
1517                    .unwrap()
1518                    .into(),
1519                }),
1520                columns: columns.into(),
1521            };
1522            let mut sql_server_desc = SqlServerTableDesc::new(sql_server_desc, vec![]).unwrap();
1523            sql_server_desc.apply_text_columns(&BTreeSet::from(["a"]));
1524
1525            // We should support decoding every datatype to a string.
1526            let relation_desc = RelationDesc::builder()
1527                .with_column("a", SqlScalarType::String.nullable(false))
1528                .finish();
1529
1530            // This decoder should shape the SQL Server Rows into Rows compatible with the RelationDesc.
1531            let decoder = sql_server_desc
1532                .decoder(&relation_desc)
1533                .expect("known valid");
1534
1535            let sql_server_row = tiberius::Row::build([(
1536                tiberius::Column::new("a".to_string(), col_type),
1537                col_data,
1538            )]);
1539            let mut mz_row = Row::default();
1540            let arena = RowArena::new();
1541            decoder
1542                .decode(&sql_server_row, &mut mz_row, &arena, None)
1543                .unwrap();
1544
1545            let str_datum = mz_row.into_element();
1546            assert!(matches!(str_datum, Datum::String(_)));
1547        }
1548
1549        use tiberius::ColumnData;
1550
1551        testcase(
1552            "bit",
1553            tiberius::ColumnType::Bit,
1554            ColumnData::Bit(Some(true)),
1555        );
1556        testcase(
1557            "bit",
1558            tiberius::ColumnType::Bit,
1559            ColumnData::Bit(Some(false)),
1560        );
1561        testcase(
1562            "tinyint",
1563            tiberius::ColumnType::Int1,
1564            ColumnData::U8(Some(33)),
1565        );
1566        testcase(
1567            "smallint",
1568            tiberius::ColumnType::Int2,
1569            ColumnData::I16(Some(101)),
1570        );
1571        testcase(
1572            "int",
1573            tiberius::ColumnType::Int4,
1574            ColumnData::I32(Some(-42)),
1575        );
1576        {
1577            let datetime = tiberius::time::DateTime::new(10, 300);
1578            testcase(
1579                "datetime",
1580                tiberius::ColumnType::Datetime,
1581                ColumnData::DateTime(Some(datetime)),
1582            );
1583        }
1584    }
1585
1586    #[mz_ore::test]
1587    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decNumberFromInt32` on OS `linux`
1588    fn smoketest_numeric_conversion() {
1589        let a = tiberius::numeric::Numeric::new_with_scale(12345, 2);
1590        let rnd = tiberius_numeric_to_mz_numeric(a);
1591        let og = mz_repr::adt::numeric::cx_datum().parse("123.45").unwrap();
1592        assert_eq!(og, rnd);
1593
1594        let a = tiberius::numeric::Numeric::new_with_scale(-99999, 5);
1595        let rnd = tiberius_numeric_to_mz_numeric(a);
1596        let og = mz_repr::adt::numeric::cx_datum().parse("-.99999").unwrap();
1597        assert_eq!(og, rnd);
1598
1599        let a = tiberius::numeric::Numeric::new_with_scale(1, 29);
1600        let rnd = tiberius_numeric_to_mz_numeric(a);
1601        let og = mz_repr::adt::numeric::cx_datum()
1602            .parse("0.00000000000000000000000000001")
1603            .unwrap();
1604        assert_eq!(og, rnd);
1605
1606        let a = tiberius::numeric::Numeric::new_with_scale(-111111111111111111, 0);
1607        let rnd = tiberius_numeric_to_mz_numeric(a);
1608        let og = mz_repr::adt::numeric::cx_datum()
1609            .parse("-111111111111111111")
1610            .unwrap();
1611        assert_eq!(og, rnd);
1612    }
1613
1614    // TODO(sql_server2): Proptest the decoder.
1615}