1use 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#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
56#[cfg_attr(any(test, feature = "proptest"), derive(Arbitrary))]
57pub struct SqlServerTableDesc {
58 pub schema_name: Arc<str>,
60 pub name: Arc<str>,
62 pub columns: Box<[SqlServerColumnDesc]>,
64 pub constraints: Vec<SqlServerTableConstraint>,
66}
67
68impl SqlServerTableDesc {
69 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 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 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 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 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#[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#[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#[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#[derive(Debug, Clone)]
269pub struct SqlServerTableRaw {
270 pub schema_name: Arc<str>,
272 pub name: Arc<str>,
274 pub capture_instance: Arc<SqlServerCaptureInstanceRaw>,
276 pub columns: Arc<[SqlServerColumnRaw]>,
278}
279
280#[derive(Debug, Clone)]
282pub struct SqlServerCaptureInstanceRaw {
283 pub name: Arc<str>,
285 pub create_date: Arc<NaiveDateTime>,
287}
288
289#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
291#[cfg_attr(any(test, feature = "proptest"), derive(Arbitrary))]
292pub struct SqlServerColumnDesc {
293 pub name: Arc<str>,
295 pub column_type: Option<SqlColumnType>,
301 pub primary_key_constraint: Option<Arc<str>>,
304 pub decode_type: SqlServerColumnDecodeType,
306 pub raw_type: Arc<str>,
310}
311
312impl SqlServerColumnDesc {
313 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 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 pub fn exclude(&mut self) {
353 self.column_type = None;
354 }
355
356 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#[derive(Debug)]
388#[allow(dead_code)]
389pub struct UnsupportedDataType {
390 column_name: String,
391 column_type: String,
392 reason: String,
393}
394
395fn parse_data_type(
400 raw: &SqlServerColumnRaw,
401) -> Result<(SqlScalarType, SqlServerColumnDecodeType), UnsupportedDataType> {
402 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 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 "real" | "float" | "double precision" => match raw.max_length {
466 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 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 (SqlScalarType::String, SqlServerColumnDecodeType::String)
496 }
497 "varchar" | "nvarchar" => {
498 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 mz_ore::soft_assert_eq_no_log!(raw.max_length, 16);
528
529 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 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 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 "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 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#[derive(Clone, Debug)]
630pub struct SqlServerColumnRaw {
631 pub name: Arc<str>,
633 pub data_type: Arc<str>,
635 pub is_nullable: bool,
637 pub max_length: i16,
647 pub precision: u8,
649 pub scale: u8,
651 pub is_computed: bool,
653}
654
655#[derive(Clone, Debug)]
657pub struct SqlServerTableConstraintRaw {
658 pub constraint_name: String,
659 pub constraint_type: String,
660 pub columns: Vec<String>,
661}
662
663#[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,
678 Numeric,
680 Xml,
682 NaiveDate,
684 NaiveTime,
686 DateTime,
688 NaiveDateTime,
690 Unsupported {
692 context: String,
694 },
695}
696
697impl SqlServerColumnDecodeType {
698 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 (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 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 let rounded = val.round_subsecs(6);
834 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 (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 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
1056fn 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 mz_repr::adt::numeric::cx_datum().scaleb(&mut numeric, &Numeric::from(-i32::from(val.scale())));
1063 numeric
1064}
1065
1066#[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 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#[derive(Debug)]
1133pub struct SqlServerRowDecoder {
1134 decoders: Vec<(Arc<str>, SqlColumnType, SqlServerColumnDecodeType, bool)>,
1135 needs_update_mask: bool,
1136}
1137
1138impl SqlServerRowDecoder {
1139 pub fn try_new(
1143 table: &SqlServerTableDesc,
1144 desc: &RelationDesc,
1145 ) -> Result<Self, SqlServerError> {
1146 let mut needs_update_mask = false;
1147 let decoders = desc
1148 .iter()
1149 .map(|(col_name, col_type)| {
1150 let sql_server_col = table
1151 .columns
1152 .iter()
1153 .find(|col| col.name.as_ref() == col_name.as_str())
1154 .ok_or_else(|| {
1155 anyhow::anyhow!("no SQL Server column with name {col_name} found")
1157 })?;
1158 let Some(sql_server_col_typ) = sql_server_col.column_type.as_ref() else {
1159 return Err(SqlServerError::ProgrammingError(format!(
1160 "programming error, {col_name} should have been exluded",
1161 )));
1162 };
1163
1164 let matches = match (&sql_server_col_typ.scalar_type, &col_type.scalar_type) {
1172 (SqlScalarType::Timestamp { .. }, SqlScalarType::Timestamp { .. })
1173 | (SqlScalarType::TimestampTz { .. }, SqlScalarType::TimestampTz { .. }) => {
1174 sql_server_col_typ.nullable == col_type.nullable
1176 }
1177 (_, _) => sql_server_col_typ == col_type,
1178 };
1179 if !matches {
1180 return Err(SqlServerError::ProgrammingError(format!(
1181 "programming error, {col_name} has mismatched type {:?} vs {:?}",
1182 sql_server_col.column_type, col_type
1183 )));
1184 }
1185
1186 let name = Arc::clone(&sql_server_col.name);
1187 let decoder = sql_server_col.decode_type.clone();
1188 let col_typ = sql_server_col_typ.clone();
1193
1194 let is_lob = is_lob_column(&col_typ.scalar_type);
1198 needs_update_mask |= is_lob;
1199
1200 Ok::<_, SqlServerError>((name, col_typ, decoder, is_lob))
1201 })
1202 .collect::<Result<_, _>>()?;
1203
1204 Ok(SqlServerRowDecoder {
1205 decoders,
1206 needs_update_mask,
1207 })
1208 }
1209
1210 pub fn decode(
1217 &self,
1218 data: &tiberius::Row,
1219 row: &mut Row,
1220 arena: &RowArena,
1221 new_data: Option<&tiberius::Row>,
1222 ) -> Result<(), SqlServerDecodeError> {
1223 let mut packer = row.packer();
1224
1225 let update_mask = new_data
1227 .filter(|_| self.needs_update_mask)
1228 .map(UpdateMask::try_from)
1229 .transpose()?;
1230 for (col_name, col_type, decoder, is_lob) in &self.decoders {
1231 let datum = if let Some((new_data, mask)) = new_data.zip(update_mask.as_ref())
1232 && *is_lob
1233 && !data_col_updated(new_data, mask, col_name)?
1234 {
1235 decoder.decode(new_data, col_name, col_type, arena)?
1240 } else {
1241 decoder.decode(data, col_name, col_type, arena)?
1242 };
1243
1244 packer.push(datum);
1245 }
1246 Ok(())
1247 }
1248
1249 pub fn included_column_names(&self) -> Vec<Arc<str>> {
1250 self.decoders
1251 .iter()
1252 .map(|decoder| Arc::clone(&decoder.0))
1253 .collect()
1254 }
1255}
1256
1257fn data_col_updated(
1260 row: &tiberius::Row,
1261 update_mask: &UpdateMask,
1262 col_name: &str,
1263) -> Result<bool, SqlServerDecodeError> {
1264 row.columns()
1265 .iter()
1266 .position(|c| c.name() == col_name)
1267 .map(|idx| update_mask.data_col_updated(idx))
1268 .ok_or_else(|| SqlServerDecodeError::MissingColumn {
1269 column_name: col_name.to_owned(),
1270 })
1271}
1272
1273fn is_lob_column(col_type: &SqlScalarType) -> bool {
1275 matches!(col_type, SqlScalarType::VarChar { max_length: None })
1276}
1277
1278#[cfg(test)]
1279mod tests {
1280 use std::collections::BTreeSet;
1281 use std::sync::Arc;
1282
1283 use chrono::NaiveDateTime;
1284 use itertools::Itertools;
1285 use mz_ore::assert_contains;
1286 use mz_ore::collections::CollectionExt;
1287 use mz_repr::adt::char::CharLength;
1288 use mz_repr::adt::numeric::NumericMaxScale;
1289 use mz_repr::adt::varchar::VarCharMaxLength;
1290 use mz_repr::{Datum, RelationDesc, Row, RowArena, SqlScalarType};
1291 use tiberius::RowTestExt;
1292
1293 use crate::desc::{
1294 SqlServerCaptureInstanceRaw, SqlServerColumnDecodeType, SqlServerColumnDesc,
1295 SqlServerTableDesc, SqlServerTableRaw, tiberius_numeric_to_mz_numeric,
1296 };
1297
1298 use super::SqlServerColumnRaw;
1299
1300 impl SqlServerColumnRaw {
1301 fn new(name: &str, data_type: &str) -> Self {
1304 SqlServerColumnRaw {
1305 name: name.into(),
1306 data_type: data_type.into(),
1307 is_nullable: false,
1308 max_length: 0,
1309 precision: 0,
1310 scale: 0,
1311 is_computed: false,
1312 }
1313 }
1314
1315 fn nullable(mut self, nullable: bool) -> Self {
1316 self.is_nullable = nullable;
1317 self
1318 }
1319
1320 fn max_length(mut self, max_length: i16) -> Self {
1321 self.max_length = max_length;
1322 self
1323 }
1324
1325 fn precision(mut self, precision: u8) -> Self {
1326 self.precision = precision;
1327 self
1328 }
1329
1330 fn scale(mut self, scale: u8) -> Self {
1331 self.scale = scale;
1332 self
1333 }
1334 }
1335
1336 #[mz_ore::test]
1337 fn smoketest_column_raw() {
1338 let raw = SqlServerColumnRaw::new("foo", "bit");
1339 let col = SqlServerColumnDesc::new(&raw);
1340
1341 assert_eq!(&*col.name, "foo");
1342 assert_eq!(col.column_type, Some(SqlScalarType::Bool.nullable(false)));
1343 assert_eq!(col.decode_type, SqlServerColumnDecodeType::Bool);
1344
1345 let raw = SqlServerColumnRaw::new("foo", "decimal")
1346 .precision(20)
1347 .scale(10);
1348 let col = SqlServerColumnDesc::new(&raw);
1349
1350 let col_type = SqlScalarType::Numeric {
1351 max_scale: Some(NumericMaxScale::try_from(10i64).expect("known valid")),
1352 }
1353 .nullable(false);
1354 assert_eq!(col.column_type, Some(col_type));
1355 assert_eq!(col.decode_type, SqlServerColumnDecodeType::Numeric);
1356 }
1357
1358 #[mz_ore::test]
1359 fn smoketest_column_raw_invalid() {
1360 let raw = SqlServerColumnRaw::new("foo", "bad_data_type");
1361 let desc = SqlServerColumnDesc::new(&raw);
1362 let SqlServerColumnDecodeType::Unsupported { context } = desc.decode_type else {
1363 panic!("unexpected decode type {desc:?}");
1364 };
1365 assert_contains!(context, "'bad_data_type' is unimplemented");
1366
1367 let raw = SqlServerColumnRaw::new("foo", "decimal")
1368 .precision(100)
1369 .scale(10);
1370 let desc = SqlServerColumnDesc::new(&raw);
1371 assert!(matches!(
1372 desc.decode_type,
1373 SqlServerColumnDecodeType::Unsupported { .. }
1374 ));
1375
1376 let raw = SqlServerColumnRaw::new("foo", "varbinary").max_length(-1);
1377 let desc = SqlServerColumnDesc::new(&raw);
1378 let SqlServerColumnDecodeType::Unsupported { context } = desc.decode_type else {
1379 panic!("unexpected decode type {desc:?}");
1380 };
1381 assert_contains!(context, "columns with unlimited size do not support CDC");
1382 }
1383
1384 #[mz_ore::test]
1385 fn smoketest_decoder() {
1386 let sql_server_columns = [
1387 SqlServerColumnRaw::new("a", "varchar").max_length(16),
1388 SqlServerColumnRaw::new("b", "int").nullable(true),
1389 SqlServerColumnRaw::new("c", "bit"),
1390 ];
1391 let sql_server_desc = SqlServerTableRaw {
1392 schema_name: "my_schema".into(),
1393 name: "my_table".into(),
1394 capture_instance: Arc::new(SqlServerCaptureInstanceRaw {
1395 name: "my_table_CT".into(),
1396 create_date: NaiveDateTime::parse_from_str(
1397 "2024-01-01 00:00:00",
1398 "%Y-%m-%d %H:%M:%S",
1399 )
1400 .unwrap()
1401 .into(),
1402 }),
1403 columns: sql_server_columns.into(),
1404 };
1405 let sql_server_desc = SqlServerTableDesc::new(sql_server_desc, vec![]).unwrap();
1406
1407 let max_length = Some(VarCharMaxLength::try_from(16).unwrap());
1408 let relation_desc = RelationDesc::builder()
1409 .with_column("a", SqlScalarType::VarChar { max_length }.nullable(false))
1410 .with_column("c", SqlScalarType::Bool.nullable(false))
1412 .with_column("b", SqlScalarType::Int32.nullable(true))
1413 .finish();
1414
1415 let decoder = sql_server_desc
1417 .decoder(&relation_desc)
1418 .expect("known valid");
1419
1420 let sql_server_columns = [
1421 tiberius::Column::new("a".to_string(), tiberius::ColumnType::BigVarChar),
1422 tiberius::Column::new("b".to_string(), tiberius::ColumnType::Int4),
1423 tiberius::Column::new("c".to_string(), tiberius::ColumnType::Bit),
1424 ];
1425
1426 let data_a = [
1427 tiberius::ColumnData::String(Some("hello world".into())),
1428 tiberius::ColumnData::I32(Some(42)),
1429 tiberius::ColumnData::Bit(Some(true)),
1430 ];
1431 let sql_server_row_a = tiberius::Row::build(
1432 sql_server_columns
1433 .iter()
1434 .cloned()
1435 .zip_eq(data_a.into_iter()),
1436 );
1437
1438 let data_b = [
1439 tiberius::ColumnData::String(Some("foo bar".into())),
1440 tiberius::ColumnData::I32(None),
1441 tiberius::ColumnData::Bit(Some(false)),
1442 ];
1443 let sql_server_row_b =
1444 tiberius::Row::build(sql_server_columns.into_iter().zip_eq(data_b.into_iter()));
1445
1446 let mut rnd_row = Row::default();
1447 let arena = RowArena::default();
1448
1449 decoder
1450 .decode(&sql_server_row_a, &mut rnd_row, &arena, None)
1451 .unwrap();
1452 assert_eq!(
1453 &rnd_row,
1454 &Row::pack_slice(&[Datum::String("hello world"), Datum::True, Datum::Int32(42)])
1455 );
1456
1457 decoder
1458 .decode(&sql_server_row_b, &mut rnd_row, &arena, None)
1459 .unwrap();
1460 assert_eq!(
1461 &rnd_row,
1462 &Row::pack_slice(&[Datum::String("foo bar"), Datum::False, Datum::Null])
1463 );
1464 }
1465
1466 #[mz_ore::test]
1467 fn decode_update_lob_before_image() {
1468 let sql_server_columns = [
1469 SqlServerColumnRaw::new("k", "int"),
1470 SqlServerColumnRaw::new("v", "varchar").max_length(-1),
1471 SqlServerColumnRaw::new("n", "varchar")
1472 .max_length(-1)
1473 .nullable(true),
1474 ];
1475 let sql_server_desc = SqlServerTableRaw {
1476 schema_name: "my_schema".into(),
1477 name: "my_table".into(),
1478 capture_instance: Arc::new(SqlServerCaptureInstanceRaw {
1479 name: "my_table_CT".into(),
1480 create_date: NaiveDateTime::parse_from_str(
1481 "2024-01-01 00:00:00",
1482 "%Y-%m-%d %H:%M:%S",
1483 )
1484 .unwrap()
1485 .into(),
1486 }),
1487 columns: sql_server_columns.into(),
1488 };
1489 let sql_server_desc = SqlServerTableDesc::new(sql_server_desc, vec![]).unwrap();
1490 let relation_desc = RelationDesc::builder()
1491 .with_column("k", SqlScalarType::Int32.nullable(false))
1492 .with_column(
1493 "v",
1494 SqlScalarType::VarChar { max_length: None }.nullable(false),
1495 )
1496 .with_column(
1497 "n",
1498 SqlScalarType::VarChar { max_length: None }.nullable(true),
1499 )
1500 .finish();
1501 let decoder = sql_server_desc
1502 .decoder(&relation_desc)
1503 .expect("known valid");
1504
1505 let cdc_row =
1508 |op: i32, mask: u8, k: i32, v: Option<&'static str>, n: Option<&'static str>| {
1509 use tiberius::{Column, ColumnData, ColumnType};
1510 let lsn = || ColumnData::Binary(Some(vec![0; 10].into()));
1511 tiberius::Row::build([
1512 (
1513 Column::new("__$start_lsn".into(), ColumnType::BigVarBin),
1514 lsn(),
1515 ),
1516 (
1517 Column::new("__$seqval".into(), ColumnType::BigVarBin),
1518 lsn(),
1519 ),
1520 (
1521 Column::new("__$operation".into(), ColumnType::Int4),
1522 ColumnData::I32(Some(op)),
1523 ),
1524 (
1525 Column::new("__$update_mask".into(), ColumnType::BigVarBin),
1526 ColumnData::Binary(Some(vec![mask].into())),
1527 ),
1528 (
1529 Column::new("k".into(), ColumnType::Int4),
1530 ColumnData::I32(Some(k)),
1531 ),
1532 (
1533 Column::new("v".into(), ColumnType::BigVarChar),
1534 ColumnData::String(v.map(Into::into)),
1535 ),
1536 (
1537 Column::new("n".into(), ColumnType::BigVarChar),
1538 ColumnData::String(n.map(Into::into)),
1539 ),
1540 ])
1541 };
1542 let arena = RowArena::new();
1543 let mut mz_row = Row::default();
1544
1545 let after = cdc_row(4, 0b001, 2, Some("v"), Some("n"));
1547 decoder
1548 .decode(
1549 &cdc_row(3, 0b001, 1, None, None),
1550 &mut mz_row,
1551 &arena,
1552 Some(&after),
1553 )
1554 .unwrap();
1555 assert_eq!(
1556 mz_row,
1557 Row::pack_slice(&[Datum::Int32(1), Datum::String("v"), Datum::String("n")])
1558 );
1559
1560 let after = cdc_row(4, 0b110, 1, Some("new"), Some("n"));
1562 decoder
1563 .decode(
1564 &cdc_row(3, 0b110, 1, Some("old"), None),
1565 &mut mz_row,
1566 &arena,
1567 Some(&after),
1568 )
1569 .unwrap();
1570 assert_eq!(
1571 mz_row,
1572 Row::pack_slice(&[Datum::Int32(1), Datum::String("old"), Datum::Null])
1573 );
1574
1575 let after = cdc_row(4, 0b010, 1, Some("new"), None);
1577 let err = decoder
1578 .decode(
1579 &cdc_row(3, 0b010, 1, None, None),
1580 &mut mz_row,
1581 &arena,
1582 Some(&after),
1583 )
1584 .unwrap_err();
1585 assert_contains!(err.to_string(), "found Null in non-nullable column");
1586 }
1587
1588 #[mz_ore::test]
1589 fn decode_legacy_char_column() {
1590 let length = Some(CharLength::try_from(10i64).expect("known valid"));
1595 let column_type = SqlScalarType::Char { length }.nullable(false);
1596 let columns = [SqlServerColumnDesc {
1597 name: "a".into(),
1598 column_type: Some(column_type.clone()),
1599 primary_key_constraint: None,
1600 decode_type: SqlServerColumnDecodeType::String,
1601 raw_type: "char".into(),
1602 }];
1603 let table_desc = SqlServerTableDesc {
1604 schema_name: "my_schema".into(),
1605 name: "my_table".into(),
1606 columns: columns.into(),
1607 constraints: vec![],
1608 };
1609 let relation_desc = RelationDesc::builder()
1610 .with_column("a", column_type)
1611 .finish();
1612 let decoder = table_desc.decoder(&relation_desc).expect("known valid");
1613
1614 let char_row = |val: &'static str| {
1615 tiberius::Row::build([(
1616 tiberius::Column::new("a".to_string(), tiberius::ColumnType::BigChar),
1617 tiberius::ColumnData::String(Some(val.into())),
1618 )])
1619 };
1620 let mut mz_row = Row::default();
1621 let arena = RowArena::default();
1622
1623 decoder
1626 .decode(&char_row("café "), &mut mz_row, &arena, None)
1627 .unwrap();
1628 assert_eq!(&mz_row, &Row::pack_slice(&[Datum::String("café ")]));
1629
1630 decoder
1632 .decode(&char_row("0123456789"), &mut mz_row, &arena, None)
1633 .unwrap();
1634 assert_eq!(&mz_row, &Row::pack_slice(&[Datum::String("0123456789")]));
1635
1636 let err = decoder
1638 .decode(&char_row("0123456789!"), &mut mz_row, &arena, None)
1639 .unwrap_err();
1640 assert_contains!(err.to_string(), "expected 10 chars found 11");
1641 }
1642
1643 #[mz_ore::test]
1644 fn smoketest_decode_to_string() {
1645 #[track_caller]
1646 fn testcase(
1647 data_type: &'static str,
1648 col_type: tiberius::ColumnType,
1649 col_data: tiberius::ColumnData<'static>,
1650 ) {
1651 let columns = [SqlServerColumnRaw::new("a", data_type)];
1652 let sql_server_desc = SqlServerTableRaw {
1653 schema_name: "my_schema".into(),
1654 name: "my_table".into(),
1655 capture_instance: Arc::new(SqlServerCaptureInstanceRaw {
1656 name: "my_table_CT".into(),
1657 create_date: NaiveDateTime::parse_from_str(
1658 "2024-01-01 00:00:00",
1659 "%Y-%m-%d %H:%M:%S",
1660 )
1661 .unwrap()
1662 .into(),
1663 }),
1664 columns: columns.into(),
1665 };
1666 let mut sql_server_desc = SqlServerTableDesc::new(sql_server_desc, vec![]).unwrap();
1667 sql_server_desc.apply_text_columns(&BTreeSet::from(["a"]));
1668
1669 let relation_desc = RelationDesc::builder()
1671 .with_column("a", SqlScalarType::String.nullable(false))
1672 .finish();
1673
1674 let decoder = sql_server_desc
1676 .decoder(&relation_desc)
1677 .expect("known valid");
1678
1679 let sql_server_row = tiberius::Row::build([(
1680 tiberius::Column::new("a".to_string(), col_type),
1681 col_data,
1682 )]);
1683 let mut mz_row = Row::default();
1684 let arena = RowArena::new();
1685 decoder
1686 .decode(&sql_server_row, &mut mz_row, &arena, None)
1687 .unwrap();
1688
1689 let str_datum = mz_row.into_element();
1690 assert!(matches!(str_datum, Datum::String(_)));
1691 }
1692
1693 use tiberius::ColumnData;
1694
1695 testcase(
1696 "bit",
1697 tiberius::ColumnType::Bit,
1698 ColumnData::Bit(Some(true)),
1699 );
1700 testcase(
1701 "bit",
1702 tiberius::ColumnType::Bit,
1703 ColumnData::Bit(Some(false)),
1704 );
1705 testcase(
1706 "tinyint",
1707 tiberius::ColumnType::Int1,
1708 ColumnData::U8(Some(33)),
1709 );
1710 testcase(
1711 "smallint",
1712 tiberius::ColumnType::Int2,
1713 ColumnData::I16(Some(101)),
1714 );
1715 testcase(
1716 "int",
1717 tiberius::ColumnType::Int4,
1718 ColumnData::I32(Some(-42)),
1719 );
1720 {
1721 let datetime = tiberius::time::DateTime::new(10, 300);
1722 testcase(
1723 "datetime",
1724 tiberius::ColumnType::Datetime,
1725 ColumnData::DateTime(Some(datetime)),
1726 );
1727 }
1728 }
1729
1730 #[mz_ore::test]
1731 #[cfg_attr(miri, ignore)] fn smoketest_numeric_conversion() {
1733 let a = tiberius::numeric::Numeric::new_with_scale(12345, 2);
1734 let rnd = tiberius_numeric_to_mz_numeric(a);
1735 let og = mz_repr::adt::numeric::cx_datum().parse("123.45").unwrap();
1736 assert_eq!(og, rnd);
1737
1738 let a = tiberius::numeric::Numeric::new_with_scale(-99999, 5);
1739 let rnd = tiberius_numeric_to_mz_numeric(a);
1740 let og = mz_repr::adt::numeric::cx_datum().parse("-.99999").unwrap();
1741 assert_eq!(og, rnd);
1742
1743 let a = tiberius::numeric::Numeric::new_with_scale(1, 29);
1744 let rnd = tiberius_numeric_to_mz_numeric(a);
1745 let og = mz_repr::adt::numeric::cx_datum()
1746 .parse("0.00000000000000000000000000001")
1747 .unwrap();
1748 assert_eq!(og, rnd);
1749
1750 let a = tiberius::numeric::Numeric::new_with_scale(-111111111111111111, 0);
1751 let rnd = tiberius_numeric_to_mz_numeric(a);
1752 let og = mz_repr::adt::numeric::cx_datum()
1753 .parse("-111111111111111111")
1754 .unwrap();
1755 assert_eq!(og, rnd);
1756 }
1757
1758 }