connectorx/macros.rs
1/// Associate physical representations to a typesystem.
2///
3/// # Example Usage
4/// ```no_run
5/// # use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime, Utc};
6/// # use connectorx::impl_typesystem;
7/// #[derive(Debug, Clone, Copy)]
8/// pub enum ArrowTypeSystem {
9/// Int32(bool),
10/// Int64(bool),
11/// UInt32(bool),
12/// UInt64(bool),
13/// Float32(bool),
14/// Float64(bool),
15/// Boolean(bool),
16/// LargeUtf8(bool),
17/// LargeBinary(bool),
18/// Date32(bool),
19/// Date64(bool),
20/// Time64(bool),
21/// DateTimeTz(bool),
22/// }
23///
24/// impl_typesystem! {
25/// system = ArrowTypeSystem,
26/// mappings = {
27/// { Int32 => i32 }
28/// { Int64 => i64 }
29/// { UInt32 => u32 }
30/// { UInt64 => u64 }
31/// { Float64 => f64 }
32/// { Float32 => f32 }
33/// { Boolean => bool }
34/// { LargeUtf8 => String }
35/// { LargeBinary => Vec<u8> }
36/// { Date32 => NaiveDate }
37/// { Date64 => NaiveDateTime }
38/// { Time64 => NaiveTime }
39/// { DateTimeTz => DateTime<Utc> }
40/// }
41/// }
42/// ```
43/// This means for the type system `ArrowTypeSystem`, it's variant `ArrowTypeSystem::Int32(false)` is corresponding to the physical type `i32` and
44/// `ArrowTypeSystem::Int32(true)` is corresponding to the physical type `Option<i32>`.
45#[macro_export]
46macro_rules! impl_typesystem {
47 (
48 system = $TS:tt,
49 mappings = {
50 $(
51 { $($V:tt)|+ => $NT:ty }
52 )*
53 }
54 ) => {
55 impl $crate::typesystem::TypeSystem for $TS {}
56
57 $(
58 impl_typesystem!(@typeassoc $TS [$($V)+], $NT);
59 )+
60
61 impl_typesystem!(@realize $TS $([ [$($V)+] => $NT ])+ );
62 };
63
64 (@typeassoc $TS:tt [$($V:tt)+], $NT:ty) => {
65 impl<'r> $crate::typesystem::TypeAssoc<$TS> for $NT {
66 fn check(ts: $TS) -> $crate::errors::Result<()> {
67 match ts {
68 $(
69 $TS::$V(false) => Ok(()),
70 )+
71 _ => fehler::throw!($crate::errors::ConnectorXError::TypeCheckFailed(format!("{:?}", ts), std::any::type_name::<$NT>()))
72 }
73 }
74 }
75
76 impl<'r> $crate::typesystem::TypeAssoc<$TS> for Option<$NT> {
77 fn check(ts: $TS) -> $crate::errors::Result<()> {
78 match ts {
79 $(
80 $TS::$V(true) => Ok(()),
81 )+
82 _ => fehler::throw!($crate::errors::ConnectorXError::TypeCheckFailed(format!("{:?}", ts), std::any::type_name::<$NT>()))
83 }
84 }
85 }
86 };
87
88 (@realize $TS:tt $([ [$($V:tt)+] => $NT:ty ])+) => {
89 impl<'r, F> $crate::typesystem::Realize<F> for $TS
90 where
91 F: $crate::typesystem::ParameterizedFunc,
92 $(F: $crate::typesystem::ParameterizedOn<$NT>,)+
93 $(F: $crate::typesystem::ParameterizedOn<Option<$NT>>,)+
94 {
95 fn realize(self) -> $crate::errors::Result<F::Function> {
96 match self {
97 $(
98 $(
99 $TS::$V(false) => Ok(F::realize::<$NT>()),
100 )+
101 $(
102 $TS::$V(true) => Ok(F::realize::<Option<$NT>>()),
103 )+
104 )+
105 }
106 }
107 }
108 };
109}
110
111/// A macro to help define a Transport.
112///
113/// # Example Usage
114/// ```no_run
115/// # use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime, Utc};
116/// # use connectorx::destinations::arrow::{ArrowDestination, ArrowDestinationError, ArrowTypeSystem};
117/// # use connectorx::sources::mssql::{FloatN, IntN, MsSQLSource, MsSQLSourceError, MsSQLTypeSystem};
118/// # use connectorx::typesystem::TypeConversion;
119/// # use connectorx::impl_transport;
120/// # use rust_decimal::Decimal;
121/// # use uuid_old::Uuid;
122/// # pub struct MsSQLArrowTransport;
123/// # #[derive(thiserror::Error, Debug)]
124/// # pub enum MsSQLArrowTransportError {
125/// # #[error(transparent)]
126/// # Source(#[from] MsSQLSourceError),
127/// # #[error(transparent)]
128/// # Destination(#[from] ArrowDestinationError),
129/// # #[error(transparent)]
130/// # ConnectorX(#[from] connectorx::errors::ConnectorXError),
131/// # }
132/// impl_transport!(
133/// name = MsSQLArrowTransport,
134/// error = MsSQLArrowTransportError,
135/// systems = MsSQLTypeSystem => ArrowTypeSystem,
136/// route = MsSQLSource => ArrowDestination,
137/// mappings = {
138/// { Tinyint[u8] => Int32[i32] | conversion auto }
139/// { Smallint[i16] => Int32[i32] | conversion auto }
140/// { Int[i32] => Int32[i32] | conversion auto }
141/// { Bigint[i64] => Int64[i64] | conversion auto }
142/// { Intn[IntN] => Int64[i64] | conversion option }
143/// { Float24[f32] => Float32[f32] | conversion auto }
144/// { Float53[f64] => Float64[f64] | conversion auto }
145/// { Floatn[FloatN] => Float64[f64] | conversion option }
146/// { Bit[bool] => Boolean[bool] | conversion auto }
147/// { Nvarchar[&'r str] => LargeUtf8[String] | conversion owned }
148/// { Varchar[&'r str] => LargeUtf8[String] | conversion none }
149/// { Nchar[&'r str] => LargeUtf8[String] | conversion none }
150/// { Char[&'r str] => LargeUtf8[String] | conversion none }
151/// { Text[&'r str] => LargeUtf8[String] | conversion none }
152/// { Ntext[&'r str] => LargeUtf8[String] | conversion none }
153/// { Binary[&'r [u8]] => LargeBinary[Vec<u8>] | conversion owned }
154/// { Varbinary[&'r [u8]] => LargeBinary[Vec<u8>] | conversion none }
155/// { Image[&'r [u8]] => LargeBinary[Vec<u8>] | conversion none }
156/// { Numeric[Decimal] => Float64[f64] | conversion option }
157/// { Decimal[Decimal] => Float64[f64] | conversion none }
158/// { Datetime[NaiveDateTime] => Date64[NaiveDateTime] | conversion auto }
159/// { Datetime2[NaiveDateTime] => Date64[NaiveDateTime] | conversion none }
160/// { Smalldatetime[NaiveDateTime] => Date64[NaiveDateTime] | conversion none }
161/// { Date[NaiveDate] => Date32[NaiveDate] | conversion auto }
162/// { Datetimeoffset[DateTime<Utc>] => DateTimeTz[DateTime<Utc>] | conversion auto }
163/// { Uniqueidentifier[Uuid] => LargeUtf8[String] | conversion option }
164/// }
165/// );
166/// # impl TypeConversion<IntN, i64> for MsSQLArrowTransport {
167/// # fn convert(val: IntN) -> i64 { val.0 }
168/// # }
169/// # impl TypeConversion<FloatN, f64> for MsSQLArrowTransport {
170/// # fn convert(val: FloatN) -> f64 { val.0 }
171/// # }
172/// # impl TypeConversion<Decimal, f64> for MsSQLArrowTransport {
173/// # fn convert(val: Decimal) -> f64 { num_traits::ToPrimitive::to_f64(&val).unwrap() }
174/// # }
175/// # impl TypeConversion<Uuid, String> for MsSQLArrowTransport {
176/// # fn convert(val: Uuid) -> String { val.to_string() }
177/// # }
178/// ```
179/// This implements a `Transport` called `MsSQLArrowTransport` that can convert types from MsSQL to Arrow.
180#[macro_export]
181macro_rules! impl_transport {
182 (
183 name = $TP:ty,
184 error = $ET:ty,
185 systems = $TSS:tt => $TSD:tt,
186 route = $S:ty => $D:ty,
187 mappings = {
188 $(
189 { $($TOKENS:tt)+ }
190 )*
191 }
192 ) => {
193 $(
194 impl_transport!(@cvt $TP, $($TOKENS)+);
195 )*
196
197 impl_transport!(@transport $TP, $ET [$TSS, $TSD] [$S, $D] $([ $($TOKENS)+ ])*);
198 };
199
200 // transport
201 (@transport $TP:ty, $ET:ty [$TSS:tt, $TSD:tt] [$S:ty, $D:ty] $([ $($TOKENS:tt)+ ])*) => {
202 impl <'tp> $crate::typesystem::Transport for $TP {
203 type TSS = $TSS;
204 type TSD = $TSD;
205 type S = $S;
206 type D = $D;
207 type Error = $ET;
208
209 impl_transport!(@cvtts [$TSS, $TSD] $([ $($TOKENS)+ ])*);
210 impl_transport!(@process [$TSS, $TSD] $([ $($TOKENS)+ ])*);
211 impl_transport!(@processor [$TSS, $TSD] $([ $($TOKENS)+ ])*, $([ $($TOKENS)+ ])*);
212 }
213 };
214
215 (@cvtts [$TSS:tt, $TSD:tt] $( [$V1:tt [$T1:ty] => $V2:tt [$T2:ty] | conversion $HOW:ident] )*) => {
216 fn convert_typesystem(ts: Self::TSS) -> $crate::errors::Result<Self::TSD> {
217 match ts {
218 $(
219 $TSS::$V1(true) => Ok($TSD::$V2(true)),
220 $TSS::$V1(false) => Ok($TSD::$V2(false)),
221 )*
222 #[allow(unreachable_patterns)]
223 _ => fehler::throw!($crate::errors::ConnectorXError::NoConversionRule(
224 format!("{:?}", ts), format!("{}", std::any::type_name::<Self::TSD>())
225 ))
226 }
227 }
228 };
229
230 (@process [$TSS:tt, $TSD:tt] $([ $V1:tt [$T1:ty] => $V2:tt [$T2:ty] | conversion $HOW:ident ])*) => {
231 fn process<'s, 'd, 'r>(
232 ts1: Self::TSS,
233 ts2: Self::TSD,
234 src: &'r mut <<Self::S as $crate::sources::Source>::Partition as $crate::sources::SourcePartition>::Parser<'s>,
235 dst: &'r mut <Self::D as $crate::destinations::Destination>::Partition<'d>,
236 ) -> Result<(), Self::Error> where Self: 'd {
237 match (ts1, ts2) {
238 $(
239 ($TSS::$V1(true), $TSD::$V2(true)) => {
240 let val: Option<$T1> = $crate::sources::PartitionParser::parse(src)?;
241 let val: Option<$T2> = <Self as TypeConversion<Option<$T1>, _>>::convert(val);
242 $crate::destinations::DestinationPartition::write(dst, val)?;
243 Ok(())
244 }
245
246 ($TSS::$V1(false), $TSD::$V2(false)) => {
247 let val: $T1 = $crate::sources::PartitionParser::parse(src)?;
248 let val: $T2 = <Self as TypeConversion<$T1, _>>::convert(val);
249 $crate::destinations::DestinationPartition::write(dst, val)?;
250 Ok(())
251 }
252 )*
253 #[allow(unreachable_patterns)]
254 _ => fehler::throw!($crate::errors::ConnectorXError::NoConversionRule(
255 format!("{:?}", ts1), format!("{:?}", ts1))
256 )
257 }
258
259 }
260 };
261
262 (@processor [$TSS:tt, $TSD:tt] $([ $V1:tt [$T1:ty] => $V2:tt [$T2:ty] | conversion $HOW:ident ])*, $([ $($TOKENS:tt)+ ])*) => {
263 fn processor<'s, 'd>(
264 ts1: Self::TSS,
265 ts2: Self::TSD,
266 ) -> $crate::errors::Result<
267 fn(
268 src: &mut <<Self::S as $crate::sources::Source>::Partition as $crate::sources::SourcePartition>::Parser<'s>,
269 dst: &mut <Self::D as $crate::destinations::Destination>::Partition<'d>,
270 ) -> Result<(), Self::Error>
271 > where Self: 'd {
272 match (ts1, ts2) {
273 $(
274 ($TSS::$V1(true), $TSD::$V2(true)) => {
275 impl_transport!(@process_func_branch true [ $($TOKENS)+ ])
276 }
277
278 ($TSS::$V1(false), $TSD::$V2(false)) => {
279 impl_transport!(@process_func_branch false [ $($TOKENS)+ ])
280 }
281 )*
282 #[allow(unreachable_patterns)]
283 _ => fehler::throw!($crate::errors::ConnectorXError::NoConversionRule(
284 format!("{:?}", ts1), format!("{:?}", ts1))
285 )
286 }
287
288 }
289 };
290
291 (@process_func_branch $OPT:ident [ $V1:tt [&$L1:lifetime $T1:ty] => $V2:tt [&$L2:lifetime $T2:ty] | conversion $HOW:ident ]) => {
292 impl_transport!(@process_func_branch $OPT &$T1, &$T2)
293 };
294 (@process_func_branch $OPT:ident [ $V1:tt [$T1:ty] => $V2:tt [&$L2:lifetime $T2:ty] | conversion $HOW:ident ]) => {
295 impl_transport!(@process_func_branch $OPT $T1, &$T2)
296 };
297 (@process_func_branch $OPT:ident [ $V1:tt [&$L1:lifetime $T1:ty] => $V2:tt [$T2:ty] | conversion $HOW:ident ]) => {
298 impl_transport!(@process_func_branch $OPT &$T1, $T2)
299 };
300 (@process_func_branch $OPT:ident [ $V1:tt [$T1:ty] => $V2:tt [$T2:ty] | conversion $HOW:ident ]) => {
301 impl_transport!(@process_func_branch $OPT $T1, $T2)
302 };
303 (@process_func_branch true $T1:ty, $T2:ty) => {
304 Ok(
305 |s: &mut _, d: &mut _| $crate::typesystem::process::<Option<$T1>, Option<$T2>, Self, Self::S, Self::D, <Self::S as $crate::sources::Source>::Error, <Self::D as $crate::destinations::Destination>::Error, Self::Error>(s, d)
306 )
307 };
308 (@process_func_branch false $T1:ty, $T2:ty) => {
309 Ok(
310 |s: &mut _, d: &mut _| $crate::typesystem::process::<$T1, $T2, Self, Self::S, Self::D, <Self::S as $crate::sources::Source>::Error, <Self::D as $crate::destinations::Destination>::Error, Self::Error>(s, d)
311 )
312 };
313
314 // TypeConversion
315 (@cvt $TP:ty, $V1:tt [$T1:ty] => $V2:tt [$T2:ty] | conversion $HOW:ident) => {
316 impl_transport!(@cvt $HOW $TP, $T1, $T2);
317 };
318 (@cvt auto $TP:ty, $T1:ty, $T2:ty) => {
319 impl<'tp, 'r> $crate::typesystem::TypeConversion<$T1, $T2> for $TP {
320 fn convert(val: $T1) -> $T2 {
321 val as _
322 }
323 }
324
325 impl_transport!(@cvt option $TP, $T1, $T2);
326 };
327 (@cvt auto_vec $TP:ty, $T1:ty, $T2:ty) => {
328 impl<'tp, 'r> $crate::typesystem::TypeConversion<$T1, $T2> for $TP {
329 fn convert(val: $T1) -> $T2 {
330 val.into_iter().map(|v| v as _).collect()
331 }
332 }
333
334 impl_transport!(@cvt option $TP, $T1, $T2);
335 };
336 (@cvt owned $TP:ty, $T1:ty, $T2:ty) => {
337 impl<'tp, 'r> $crate::typesystem::TypeConversion<$T1, $T2> for $TP {
338 fn convert(val: $T1) -> $T2 {
339 val.to_owned()
340 }
341 }
342
343 impl_transport!(@cvt option $TP, $T1, $T2);
344 };
345 (@cvt option $TP:ty, $T1:ty, $T2:ty) => {
346 impl<'tp, 'r> $crate::typesystem::TypeConversion<Option<$T1>, Option<$T2>> for $TP {
347 fn convert(val: Option<$T1>) -> Option<$T2> {
348 val.map(Self::convert)
349 }
350 }
351 };
352 (@cvt none $TP:ty, $T1:ty, $T2:ty) => {};
353}