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

risingwave_connector/parser/
sql_server.rs
