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)>,
1135}
1136
1137impl SqlServerRowDecoder {
1138 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 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 let matches = match (&sql_server_col_typ.scalar_type, &col_type.scalar_type) {
1170 (SqlScalarType::Timestamp { .. }, SqlScalarType::Timestamp { .. })
1171 | (SqlScalarType::TimestampTz { .. }, SqlScalarType::TimestampTz { .. }) => {
1172 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 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 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 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 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 .with_column("c", SqlScalarType::Bool.nullable(false))
1390 .with_column("b", SqlScalarType::Int32.nullable(true))
1391 .finish();
1392
1393 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 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 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 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 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 let relation_desc = RelationDesc::builder()
1527 .with_column("a", SqlScalarType::String.nullable(false))
1528 .finish();
1529
1530 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)] 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 }