risingwave_connector/parser/
sql_server.rs

1// Copyright 2024 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::HashSet;
16use std::str::FromStr;
17use std::sync::LazyLock;
18
19use anyhow::{Context, bail};
20use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
21use risingwave_common::catalog::Schema;
22use risingwave_common::log::LogSuppressor;
23use risingwave_common::row::OwnedRow;
24use risingwave_common::types::{
25    DataType, Date, Datum, Decimal, ScalarImpl, Time, Timestamp, Timestamptz,
26};
27use rust_decimal::Decimal as RustDecimal;
28use thiserror_ext::AsReport;
29use tiberius::Row;
30use tiberius::xml::XmlData;
31use uuid::Uuid;
32
33use crate::parser::utils::log_error;
34
35static LOG_SUPPRESSOR: LazyLock<LogSuppressor> = LazyLock::new(LogSuppressor::default);
36
37pub fn sql_server_row_to_owned_row(row: &mut Row, schema: &Schema) -> OwnedRow {
38    let money_fields = sql_server_money_fields(row);
39    let mut datums = Vec::with_capacity(schema.fields.len());
40    for (i, rw_field) in schema.fields.iter().enumerate() {
41        let name = rw_field.name.as_str();
42        let datum = match sql_server_cell_to_rw_datum(
43            row,
44            i,
45            name,
46            &rw_field.data_type,
47            money_fields.contains(name),
48        ) {
49            Ok(datum) => datum,
50            Err(err) => {
51                log_error!(name, err, "parse column failed");
52                None
53            }
54        };
55        datums.push(datum);
56    }
57    OwnedRow::new(datums)
58}
59
60/// Decode primary-key columns strictly while preserving the legacy lenient behavior for all
61/// other columns in a SQL Server CDC snapshot row.
62pub fn sql_server_row_to_owned_row_with_strict_pk(
63    row: &mut Row,
64    schema: &Schema,
65    pk_indices: &[usize],
66) -> anyhow::Result<OwnedRow> {
67    let money_fields = sql_server_money_fields(row);
68    super::decode_row_with_strict_pk(
69        "SQL Server",
70        schema,
71        pk_indices,
72        |index, field| {
73            sql_server_cell_to_rw_datum(
74                row,
75                index,
76                &field.name,
77                &field.data_type,
78                money_fields.contains(&field.name),
79            )
80        },
81        |name, err| log_error!(name, err, "parse column failed"),
82    )
83}
84
85fn sql_server_money_fields(row: &Row) -> HashSet<String> {
86    let mut money_fields = HashSet::new();
87    // Special handling of the money field, as the third-party library Tiberius converts the money type to i64.
88    for (column, _) in row.cells() {
89        if column.column_type() == tiberius::ColumnType::Money {
90            money_fields.insert(column.name().to_owned());
91        }
92    }
93    money_fields
94}
95
96fn sql_server_cell_to_rw_datum(
97    row: &Row,
98    index: usize,
99    name: &str,
100    data_type: &DataType,
101    is_money: bool,
102) -> anyhow::Result<Datum> {
103    if is_money {
104        return row
105            .try_get::<i64, usize>(index)
106            .with_context(|| format!("failed to decode SQL Server money column `{name}`"))?
107            .map(|value| try_convert_money_i64_to_type(value, data_type))
108            .transpose();
109    }
110
111    Ok(row
112        .try_get::<ScalarImplTiberiusWrapper, usize>(index)
113        .with_context(|| format!("failed to decode SQL Server snapshot column `{name}`"))?
114        .map(|datum| datum.0)
115        .map(|scalar| coerce_scalar_to_target_type(scalar, data_type)))
116}
117
118fn coerce_scalar_to_target_type(scalar: ScalarImpl, target_type: &DataType) -> ScalarImpl {
119    match (scalar, target_type) {
120        // SQL Server validator allows integer upcast (e.g. `int` -> `BIGINT`).
121        // Coerce snapshot values to the target RW type to keep validation and execution consistent.
122        (ScalarImpl::Int16(v), DataType::Int32) => ScalarImpl::Int32(v as i32),
123        (ScalarImpl::Int16(v), DataType::Int64) => ScalarImpl::Int64(v as i64),
124        (ScalarImpl::Int32(v), DataType::Int64) => ScalarImpl::Int64(v as i64),
125        // SQL Server `real` may map to `FLOAT` in RW validator.
126        (ScalarImpl::Float32(v), DataType::Float64) => ScalarImpl::Float64((v.0 as f64).into()),
127        (scalar, _) => scalar,
128    }
129}
130
131fn try_convert_money_i64_to_type(value: i64, data_type: &DataType) -> anyhow::Result<ScalarImpl> {
132    match data_type {
133        DataType::Decimal => Ok(ScalarImpl::Decimal(
134            Decimal::from(value) / Decimal::from_str("10000").unwrap(),
135        )),
136        _ => bail!("conversion of SQL Server money to {data_type} is not supported"),
137    }
138}
139
140#[cfg(test)]
141mod tests {
142    use risingwave_common::types::F32;
143
144    use super::*;
145
146    #[test]
147    fn test_integer_upcast_coercion() {
148        let v = coerce_scalar_to_target_type(ScalarImpl::Int32(7), &DataType::Int64);
149        assert_eq!(v, ScalarImpl::Int64(7));
150
151        let v = coerce_scalar_to_target_type(ScalarImpl::Int16(7), &DataType::Int32);
152        assert_eq!(v, ScalarImpl::Int32(7));
153
154        let v = coerce_scalar_to_target_type(ScalarImpl::Int16(7), &DataType::Int64);
155        assert_eq!(v, ScalarImpl::Int64(7));
156    }
157
158    #[test]
159    fn test_float_upcast_coercion() {
160        let v =
161            coerce_scalar_to_target_type(ScalarImpl::Float32(F32::from(1.25)), &DataType::Float64);
162        assert_eq!(v, ScalarImpl::Float64(1.25.into()));
163    }
164
165    #[test]
166    fn test_non_upcast_keeps_original() {
167        let v = coerce_scalar_to_target_type(ScalarImpl::Int32(7), &DataType::Int32);
168        assert_eq!(v, ScalarImpl::Int32(7));
169    }
170}
171macro_rules! impl_tiberius_wrapper {
172    ($wrapper_name:ident, $variant_name:ident) => {
173        pub struct $wrapper_name($variant_name);
174
175        impl From<$variant_name> for $wrapper_name {
176            fn from(value: $variant_name) -> Self {
177                Self(value)
178            }
179        }
180    };
181}
182
183impl_tiberius_wrapper!(ScalarImplTiberiusWrapper, ScalarImpl);
184impl_tiberius_wrapper!(TimeTiberiusWrapper, Time);
185impl_tiberius_wrapper!(DateTiberiusWrapper, Date);
186impl_tiberius_wrapper!(TimestampTiberiusWrapper, Timestamp);
187impl_tiberius_wrapper!(TimestamptzTiberiusWrapper, Timestamptz);
188impl_tiberius_wrapper!(DecimalTiberiusWrapper, Decimal);
189
190macro_rules! impl_chrono_tiberius_wrapper {
191    ($wrapper_name:ident, $variant_name:ident, $chrono:ty) => {
192        impl<'a> tiberius::IntoSql<'a> for $wrapper_name {
193            fn into_sql(self) -> tiberius::ColumnData<'a> {
194                self.0.0.into_sql()
195            }
196        }
197
198        impl<'a> tiberius::FromSql<'a> for $wrapper_name {
199            fn from_sql(
200                value: &'a tiberius::ColumnData<'static>,
201            ) -> tiberius::Result<Option<Self>> {
202                let instant = <$chrono>::from_sql(value)?;
203                let time = instant.map($variant_name::from).map($wrapper_name::from);
204                tiberius::Result::Ok(time)
205            }
206        }
207    };
208}
209
210impl_chrono_tiberius_wrapper!(TimeTiberiusWrapper, Time, NaiveTime);
211impl_chrono_tiberius_wrapper!(DateTiberiusWrapper, Date, NaiveDate);
212impl_chrono_tiberius_wrapper!(TimestampTiberiusWrapper, Timestamp, NaiveDateTime);
213
214impl<'a> tiberius::IntoSql<'a> for DecimalTiberiusWrapper {
215    fn into_sql(self) -> tiberius::ColumnData<'a> {
216        match self.0 {
217            Decimal::Normalized(d) => d.into_sql(),
218            Decimal::NaN => tiberius::ColumnData::Numeric(None),
219            Decimal::PositiveInf => tiberius::ColumnData::Numeric(None),
220            Decimal::NegativeInf => tiberius::ColumnData::Numeric(None),
221        }
222    }
223}
224
225impl<'a> tiberius::FromSql<'a> for DecimalTiberiusWrapper {
226    // TODO(kexiang): will sql server have inf/-inf/nan for decimal?
227    fn from_sql(value: &'a tiberius::ColumnData<'static>) -> tiberius::Result<Option<Self>> {
228        tiberius::Result::Ok(
229            RustDecimal::from_sql(value)?
230                .map(Decimal::Normalized)
231                .map(DecimalTiberiusWrapper::from),
232        )
233    }
234}
235
236impl<'a> tiberius::IntoSql<'a> for TimestamptzTiberiusWrapper {
237    fn into_sql(self) -> tiberius::ColumnData<'a> {
238        self.0.to_datetime_utc().into_sql()
239    }
240}
241
242impl<'a> tiberius::FromSql<'a> for TimestamptzTiberiusWrapper {
243    fn from_sql(value: &'a tiberius::ColumnData<'static>) -> tiberius::Result<Option<Self>> {
244        let instant = time::OffsetDateTime::from_sql(value)?;
245        instant
246            .map(|instant| {
247                let timestamptz = instant
248                    .unix_timestamp_nanos()
249                    .checked_div(1000)
250                    .and_then(|micros| i64::try_from(micros).ok())
251                    .and_then(Timestamptz::from_micros)
252                    .ok_or_else(|| {
253                        tiberius::error::Error::Conversion(
254                            "datetimeoffset is out of range for RisingWave timestamptz".into(),
255                        )
256                    })?;
257                Ok(TimestamptzTiberiusWrapper::from(timestamptz))
258            })
259            .transpose()
260    }
261}
262
263/// The following table shows the mapping between Rust types and Sql Server types in tiberius.
264/// |Rust Type|Sql Server Type|
265/// |`u8`|`tinyint`|
266/// |`i16`|`smallint`|
267/// |`i32`|`int`|
268/// |`i64`|`bigint`|
269/// |`f32`|`float(24)`|
270/// |`f64`|`float(53)`|
271/// |`bool`|`bit`|
272/// |`String`/`&str`|`nvarchar`/`varchar`/`nchar`/`char`/`ntext`/`text`|
273/// |`Vec<u8>`/`&[u8]`|`binary`/`varbinary`/`image`|
274/// |[`Uuid`]|`uniqueidentifier`|
275/// |[`Numeric`]|`numeric`/`decimal`|
276/// |[`Decimal`] (with feature flag `rust_decimal`)|`numeric`/`decimal`|
277/// |[`XmlData`]|`xml`|
278/// |[`NaiveDateTime`] (with feature flag `chrono`)|`datetime`/`datetime2`/`smalldatetime`|
279/// |[`NaiveDate`] (with feature flag `chrono`)|`date`|
280/// |[`NaiveTime`] (with feature flag `chrono`)|`time`|
281/// |[`DateTime`] (with feature flag `chrono`)|`datetimeoffset`|
282///
283/// See the [`time`] module for more information about the date and time structs.
284///
285/// [`Row#get`]: struct.Row.html#method.get
286/// [`Row#try_get`]: struct.Row.html#method.try_get
287/// [`time`]: time/index.html
288/// [`Uuid`]: struct.Uuid.html
289/// [`Numeric`]: numeric/struct.Numeric.html
290/// [`Decimal`]: numeric/struct.Decimal.html
291/// [`XmlData`]: xml/struct.XmlData.html
292/// [`NaiveDateTime`]: time/chrono/struct.NaiveDateTime.html
293/// [`NaiveDate`]: time/chrono/struct.NaiveDate.html
294/// [`NaiveTime`]: time/chrono/struct.NaiveTime.html
295/// [`DateTime`]: time/chrono/struct.DateTime.html
296impl<'a> tiberius::FromSql<'a> for ScalarImplTiberiusWrapper {
297    fn from_sql(value: &'a tiberius::ColumnData<'static>) -> tiberius::Result<Option<Self>> {
298        Ok(match &value {
299            tiberius::ColumnData::U8(_) => u8::from_sql(value)?
300                .map(|v| ScalarImplTiberiusWrapper::from(ScalarImpl::from(v as i16))),
301            tiberius::ColumnData::I16(_) => i16::from_sql(value)?
302                .map(ScalarImpl::from)
303                .map(ScalarImplTiberiusWrapper::from),
304            tiberius::ColumnData::I32(_) => i32::from_sql(value)?
305                .map(ScalarImpl::from)
306                .map(ScalarImplTiberiusWrapper::from),
307            tiberius::ColumnData::I64(_) => i64::from_sql(value)?
308                .map(ScalarImpl::from)
309                .map(ScalarImplTiberiusWrapper::from),
310            tiberius::ColumnData::F32(_) => f32::from_sql(value)?
311                .map(ScalarImpl::from)
312                .map(ScalarImplTiberiusWrapper::from),
313            tiberius::ColumnData::F64(_) => f64::from_sql(value)?
314                .map(ScalarImpl::from)
315                .map(ScalarImplTiberiusWrapper::from),
316            tiberius::ColumnData::Bit(_) => bool::from_sql(value)?
317                .map(ScalarImpl::from)
318                .map(ScalarImplTiberiusWrapper::from),
319            tiberius::ColumnData::String(_) => <&str>::from_sql(value)?
320                .map(ScalarImpl::from)
321                .map(ScalarImplTiberiusWrapper::from),
322            tiberius::ColumnData::Numeric(_) => DecimalTiberiusWrapper::from_sql(value)?
323                .map(|w| ScalarImpl::from(w.0))
324                .map(ScalarImplTiberiusWrapper::from),
325            tiberius::ColumnData::DateTime(_)
326            | tiberius::ColumnData::DateTime2(_)
327            | tiberius::ColumnData::SmallDateTime(_) => TimestampTiberiusWrapper::from_sql(value)?
328                .map(|w| ScalarImpl::from(w.0))
329                .map(ScalarImplTiberiusWrapper::from),
330            tiberius::ColumnData::Time(_) => TimeTiberiusWrapper::from_sql(value)?
331                .map(|w| ScalarImpl::from(w.0))
332                .map(ScalarImplTiberiusWrapper::from),
333            tiberius::ColumnData::Date(_) => DateTiberiusWrapper::from_sql(value)?
334                .map(|w| ScalarImpl::from(w.0))
335                .map(ScalarImplTiberiusWrapper::from),
336            tiberius::ColumnData::DateTimeOffset(_) => TimestamptzTiberiusWrapper::from_sql(value)?
337                .map(|w| ScalarImpl::from(w.0))
338                .map(ScalarImplTiberiusWrapper::from),
339            tiberius::ColumnData::Binary(_) => <&[u8]>::from_sql(value)?
340                .map(ScalarImpl::from)
341                .map(ScalarImplTiberiusWrapper::from),
342            tiberius::ColumnData::Guid(_) => <Uuid>::from_sql(value)?
343                .map(|uuid| uuid.to_string().to_uppercase())
344                .map(ScalarImpl::from)
345                .map(ScalarImplTiberiusWrapper::from),
346            tiberius::ColumnData::Xml(_) => <&XmlData>::from_sql(value)?
347                .map(|xml| xml.clone().into_string())
348                .map(ScalarImpl::from)
349                .map(ScalarImplTiberiusWrapper::from),
350        })
351    }
352}
353
354/// The following table shows the mapping between Rust types and Sql Server types in tiberius.
355/// |Rust type|Sql Server type|
356/// |--------|--------|
357/// |`u8`|`tinyint`|
358/// |`i16`|`smallint`|
359/// |`i32`|`int`|
360/// |`i64`|`bigint`|
361/// |`f32`|`float(24)`|
362/// |`f64`|`float(53)`|
363/// |`bool`|`bit`|
364/// |`String`/`&str` (< 4000 characters)|`nvarchar(4000)`|
365/// |`String`/`&str`|`nvarchar(max)`|
366/// |`Vec<u8>`/`&[u8]` (< 8000 bytes)|`varbinary(8000)`|
367/// |`Vec<u8>`/`&[u8]`|`varbinary(max)`|
368/// |[`Uuid`]|`uniqueidentifier`|
369/// |[`Numeric`]|`numeric`/`decimal`|
370/// |[`Decimal`] (with feature flag `rust_decimal`)|`numeric`/`decimal`|
371/// |[`BigDecimal`] (with feature flag `bigdecimal`)|`numeric`/`decimal`|
372/// |[`XmlData`]|`xml`|
373/// |[`NaiveDate`] (with `chrono` feature, TDS 7.3 >)|`date`|
374/// |[`NaiveTime`] (with `chrono` feature, TDS 7.3 >)|`time`|
375/// |[`DateTime`] (with `chrono` feature, TDS 7.3 >)|`datetimeoffset`|
376/// |[`NaiveDateTime`] (with `chrono` feature, TDS 7.3 >)|`datetime2`|
377/// |[`NaiveDateTime`] (with `chrono` feature, TDS 7.2)|`datetime`|
378///
379/// It is possible to use some of the types to write into columns that are not
380/// of the same type. For example on systems following the TDS 7.3 standard (SQL
381/// Server 2008 and later), the chrono type `NaiveDateTime` can also be used to
382/// write to `datetime`, `datetime2` and `smalldatetime` columns. All string
383/// types can also be used with `ntext`, `text`, `varchar`, `nchar` and `char`
384/// columns. All binary types can also be used with `binary` and `image`
385/// columns.
386///
387/// See the [`time`] module for more information about the date and time structs.
388///
389/// [`Client#query`]: struct.Client.html#method.query
390/// [`Client#execute`]: struct.Client.html#method.execute
391/// [`time`]: time/index.html
392/// [`Uuid`]: struct.Uuid.html
393/// [`Numeric`]: numeric/struct.Numeric.html
394/// [`Decimal`]: numeric/struct.Decimal.html
395/// [`BigDecimal`]: numeric/struct.BigDecimal.html
396/// [`XmlData`]: xml/struct.XmlData.html
397/// [`NaiveDateTime`]: time/chrono/struct.NaiveDateTime.html
398/// [`NaiveDate`]: time/chrono/struct.NaiveDate.html
399/// [`NaiveTime`]: time/chrono/struct.NaiveTime.html
400/// [`DateTime`]: time/chrono/struct.DateTime.html
401impl<'a> tiberius::IntoSql<'a> for ScalarImplTiberiusWrapper {
402    fn into_sql(self) -> tiberius::ColumnData<'a> {
403        match self.0 {
404            ScalarImpl::Int16(v) => v.into_sql(),
405            ScalarImpl::Int32(v) => v.into_sql(),
406            ScalarImpl::Int64(v) => v.into_sql(),
407            ScalarImpl::Float32(v) => v.0.into_sql(),
408            ScalarImpl::Float64(v) => v.0.into_sql(),
409            ScalarImpl::Bool(v) => v.into_sql(),
410            ScalarImpl::Decimal(v) => DecimalTiberiusWrapper::from(v).into_sql(),
411            ScalarImpl::Date(v) => DateTiberiusWrapper::from(v).into_sql(),
412            ScalarImpl::Timestamp(v) => TimestampTiberiusWrapper::from(v).into_sql(),
413            ScalarImpl::Timestamptz(v) => TimestamptzTiberiusWrapper::from(v).into_sql(),
414            ScalarImpl::Time(v) => TimeTiberiusWrapper::from(v).into_sql(),
415            ScalarImpl::Bytea(v) => {
416                let value: Vec<u8> = (*v).to_vec();
417                value.into_sql()
418            }
419            ScalarImpl::Utf8(v) => String::from(v).into_sql(),
420            value => {
421                // Serial, Interval, Jsonb, Int256, Struct, List are not supported yet
422                unimplemented!("the sql server decoding for {:?} is unsupported", value);
423            }
424        }
425    }
426}
risingwave_connector/parser/sql_server.rs

risingwave_connector/parser/
sql_server.rs
