connectorx/
macros.rs

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

connectorx/
macros.rs
