sfu-db · holicc · May 15, 2025 · May 15, 2025 · May 15, 2025
diff --git a/connectorx/src/destinations/arrow/arrow_assoc.rs b/connectorx/src/destinations/arrow/arrow_assoc.rs
@@ -2,17 +2,18 @@ use super::{
     errors::{ArrowDestinationError, Result},
     typesystem::{DateTimeWrapperMicro, NaiveDateTimeWrapperMicro, NaiveTimeWrapperMicro},
 };
-use crate::constants::SECONDS_IN_DAY;
+use crate::{constants::SECONDS_IN_DAY, utils::decimal_to_i128};
 use arrow::array::{
-    ArrayBuilder, BooleanBuilder, Date32Builder, Float32Builder, Float64Builder, Int16Builder,
-    Int32Builder, Int64Builder, LargeBinaryBuilder, LargeListBuilder, StringBuilder,
+    ArrayBuilder, BooleanBuilder, Date32Builder, Decimal128Builder, Float32Builder, Float64Builder,
+    Int16Builder, Int32Builder, Int64Builder, LargeBinaryBuilder, LargeListBuilder, StringBuilder,
     Time64MicrosecondBuilder, Time64NanosecondBuilder, TimestampMicrosecondBuilder,
     TimestampNanosecondBuilder, UInt16Builder, UInt32Builder, UInt64Builder,
 };
 use arrow::datatypes::Field;
 use arrow::datatypes::{DataType as ArrowDataType, TimeUnit};
 use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime, Timelike, Utc};
 use fehler::throws;
+use rust_decimal::Decimal;
 
 /// Associate arrow builder with native type
 pub trait ArrowAssoc {
@@ -71,6 +72,51 @@ impl_arrow_assoc!(f32, ArrowDataType::Float32, Float32Builder);
 impl_arrow_assoc!(f64, ArrowDataType::Float64, Float64Builder);
 impl_arrow_assoc!(bool, ArrowDataType::Boolean, BooleanBuilder);
 
+const DEFAULT_ARROW_DECIMAL_PRECISION: u8 = 38;
+const DEFAULT_ARROW_DECIMAL_SCALE: i8 = 10;
+const DEFAULT_ARROW_DECIMAL: ArrowDataType =
+    ArrowDataType::Decimal128(DEFAULT_ARROW_DECIMAL_PRECISION, DEFAULT_ARROW_DECIMAL_SCALE);
+
+impl ArrowAssoc for Decimal {
+    type Builder = Decimal128Builder;
+
+    fn builder(nrows: usize) -> Self::Builder {
+        Decimal128Builder::with_capacity(nrows).with_data_type(DEFAULT_ARROW_DECIMAL)
+    }
+
+    fn append(builder: &mut Self::Builder, value: Self) -> Result<()> {
+        builder.append_value(decimal_to_i128(value, DEFAULT_ARROW_DECIMAL_SCALE as u32)?);
+        Ok(())
+    }
+
+    fn field(header: &str) -> Field {
+        Field::new(header, DEFAULT_ARROW_DECIMAL, false)
+    }
+}
+
+impl ArrowAssoc for Option<Decimal> {
+    type Builder = Decimal128Builder;
+
+    fn builder(nrows: usize) -> Self::Builder {
+        Decimal128Builder::with_capacity(nrows).with_data_type(DEFAULT_ARROW_DECIMAL)
+    }
+
+    fn append(builder: &mut Self::Builder, value: Self) -> Result<()> {
+        match value {
+            Some(v) => builder.append_option(Some(decimal_to_i128(
+                v,
+                DEFAULT_ARROW_DECIMAL_SCALE as u32,
+            )?)),
+            None => builder.append_null(),
+        }
+        Ok(())
+    }
+
+    fn field(header: &str) -> Field {
+        Field::new(header, DEFAULT_ARROW_DECIMAL, true)
+    }
+}
+
 impl ArrowAssoc for &str {
     type Builder = StringBuilder;
 
@@ -486,6 +532,93 @@ impl ArrowAssoc for Vec<u8> {
     }
 }
 
+impl ArrowAssoc for Option<Vec<Option<Decimal>>> {
+    type Builder = LargeListBuilder<Decimal128Builder>;
+
+    fn builder(nrows: usize) -> Self::Builder {
+        LargeListBuilder::with_capacity(
+            Decimal128Builder::with_capacity(nrows).with_data_type(DEFAULT_ARROW_DECIMAL),
+            nrows,
+        )
+    }
+
+    fn append(builder: &mut Self::Builder, value: Self) -> Result<()> {
+        match value {
+            Some(vals) => {
+                let mut list = vec![];
+
+                for val in vals {
+                    match val {
+                        Some(v) => {
+                            list.push(Some(decimal_to_i128(
+                                v,
+                                DEFAULT_ARROW_DECIMAL_SCALE as u32,
+                            )?));
+                        }
+                        None => list.push(None),
+                    }
+                }
+
+                builder.append_value(list);
+            }
+            None => builder.append_null(),
+        };
+        Ok(())
+    }
+
+    fn field(header: &str) -> Field {
+        Field::new(
+            header,
+            ArrowDataType::LargeList(std::sync::Arc::new(Field::new_list_field(
+                DEFAULT_ARROW_DECIMAL,
+                true,
+            ))),
+            true,
+        )
+    }
+}
+
+impl ArrowAssoc for Vec<Option<Decimal>> {
+    type Builder = LargeListBuilder<Decimal128Builder>;
+
+    fn builder(nrows: usize) -> Self::Builder {
+        LargeListBuilder::with_capacity(
+            Decimal128Builder::with_capacity(nrows).with_data_type(DEFAULT_ARROW_DECIMAL),
+            nrows,
+        )
+    }
+
+    fn append(builder: &mut Self::Builder, vals: Self) -> Result<()> {
+        let mut list = vec![];
+
+        for val in vals {
+            match val {
+                Some(v) => {
+                    list.push(Some(decimal_to_i128(
+                        v,
+                        DEFAULT_ARROW_DECIMAL_SCALE as u32,
+                    )?));
+                }
+                None => list.push(None),
+            }
+        }
+
+        builder.append_value(list);
+        Ok(())
+    }
+
+    fn field(header: &str) -> Field {
+        Field::new(
+            header,
+            ArrowDataType::LargeList(std::sync::Arc::new(Field::new_list_field(
+                DEFAULT_ARROW_DECIMAL,
+                false,
+            ))),
+            false,
+        )
+    }
+}
+
 macro_rules! impl_arrow_array_assoc {
     ($T:ty, $AT:expr, $B:ident) => {
         impl ArrowAssoc for $T {

diff --git a/connectorx/src/destinations/arrow/typesystem.rs b/connectorx/src/destinations/arrow/typesystem.rs
@@ -1,5 +1,6 @@
 use crate::impl_typesystem;
 use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime, Utc};
+use rust_decimal::Decimal;
 
 #[derive(Debug, Clone, Copy)]
 pub struct DateTimeWrapperMicro(pub DateTime<Utc>);
@@ -20,6 +21,7 @@ pub enum ArrowTypeSystem {
     UInt64(bool),
     Float32(bool),
     Float64(bool),
+    Decimal(bool),
     Boolean(bool),
     LargeUtf8(bool),
     LargeBinary(bool),
@@ -40,6 +42,7 @@ pub enum ArrowTypeSystem {
     UInt64Array(bool),
     Float32Array(bool),
     Float64Array(bool),
+    DecimalArray(bool),
 }
 
 impl_typesystem! {
@@ -53,6 +56,7 @@ impl_typesystem! {
         { UInt64          => u64                       }
         { Float32         => f32                       }
         { Float64         => f64                       }
+        { Decimal         => Decimal                   }
         { Boolean         => bool                      }
         { LargeUtf8       => String                    }
         { LargeBinary     => Vec<u8>                   }
@@ -73,5 +77,6 @@ impl_typesystem! {
         { UInt64Array     => Vec<Option<u64>>          }
         { Float32Array    => Vec<Option<f32>>          }
         { Float64Array    => Vec<Option<f64>>          }
+        { DecimalArray    => Vec<Option<Decimal>>      }
     }
 }
diff --git a/connectorx/src/destinations/arrowstream/arrow_assoc.rs b/connectorx/src/destinations/arrowstream/arrow_assoc.rs
@@ -1,15 +1,21 @@
 use super::errors::{ArrowDestinationError, Result};
 use crate::constants::SECONDS_IN_DAY;
+use crate::utils::decimal_to_i128;
 use arrow::array::{
-    ArrayBuilder, BooleanBuilder, Date32Builder, Date64Builder, Float32Builder, Float64Builder,
-    Int32Builder, Int64Builder, LargeBinaryBuilder, StringBuilder, Time64NanosecondBuilder,
-    TimestampNanosecondBuilder, UInt32Builder, UInt64Builder,
+    ArrayBuilder, BooleanBuilder, Date32Builder, Date64Builder, Decimal128Builder, Float32Builder,
+    Float64Builder, Int32Builder, Int64Builder, LargeBinaryBuilder, StringBuilder,
+    Time64NanosecondBuilder, TimestampNanosecondBuilder, UInt32Builder, UInt64Builder,
 };
 use arrow::datatypes::Field;
 use arrow::datatypes::{DataType as ArrowDataType, TimeUnit};
 use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime, Timelike, Utc};
 use fehler::throws;
+use rust_decimal::Decimal;
 
+const DEFAULT_ARROW_DECIMAL_PRECISION: u8 = 38;
+const DEFAULT_ARROW_DECIMAL_SCALE: i8 = 10;
+const DEFAULT_ARROW_DECIMAL: ArrowDataType =
+    ArrowDataType::Decimal128(DEFAULT_ARROW_DECIMAL_PRECISION, DEFAULT_ARROW_DECIMAL_SCALE);
 /// Associate arrow builder with native type
 pub trait ArrowAssoc {
     type Builder: ArrayBuilder + Send;
@@ -65,6 +71,46 @@ impl_arrow_assoc!(f32, ArrowDataType::Float32, Float32Builder);
 impl_arrow_assoc!(f64, ArrowDataType::Float64, Float64Builder);
 impl_arrow_assoc!(bool, ArrowDataType::Boolean, BooleanBuilder);
 
+impl ArrowAssoc for Decimal {
+    type Builder = Decimal128Builder;
+
+    fn builder(nrows: usize) -> Self::Builder {
+        Decimal128Builder::with_capacity(nrows).with_data_type(DEFAULT_ARROW_DECIMAL)
+    }
+
+    fn append(builder: &mut Self::Builder, value: Self) -> Result<()> {
+        builder.append_value(decimal_to_i128(value, DEFAULT_ARROW_DECIMAL_SCALE as u32)?);
+        Ok(())
+    }
+
+    fn field(header: &str) -> Field {
+        Field::new(header, DEFAULT_ARROW_DECIMAL, false)
+    }
+}
+
+impl ArrowAssoc for Option<Decimal> {
+    type Builder = Decimal128Builder;
+
+    fn builder(nrows: usize) -> Self::Builder {
+        Decimal128Builder::with_capacity(nrows).with_data_type(DEFAULT_ARROW_DECIMAL)
+    }
+
+    fn append(builder: &mut Self::Builder, value: Self) -> Result<()> {
+        match value {
+            Some(v) => builder.append_option(Some(decimal_to_i128(
+                v,
+                DEFAULT_ARROW_DECIMAL_SCALE as u32,
+            )?)),
+            None => builder.append_null(),
+        }
+        Ok(())
+    }
+
+    fn field(header: &str) -> Field {
+        Field::new(header, DEFAULT_ARROW_DECIMAL, true)
+    }
+}
+
 impl ArrowAssoc for &str {
     type Builder = StringBuilder;
 

diff --git a/connectorx/src/destinations/arrowstream/typesystem.rs b/connectorx/src/destinations/arrowstream/typesystem.rs
@@ -1,5 +1,6 @@
 use crate::impl_typesystem;
 use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime, Utc};
+use rust_decimal::Decimal;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
 pub enum ArrowTypeSystem {
@@ -9,6 +10,7 @@ pub enum ArrowTypeSystem {
     UInt64(bool),
     Float32(bool),
     Float64(bool),
+    Decimal(bool),
     Boolean(bool),
     LargeUtf8(bool),
     LargeBinary(bool),
@@ -27,6 +29,7 @@ impl_typesystem! {
         { UInt64     => u64           }
         { Float64    => f64           }
         { Float32    => f32           }
+        { Decimal    => Decimal       }
         { Boolean    => bool          }
         { LargeUtf8  => String        }
         { LargeBinary => Vec<u8>      }

diff --git a/connectorx/src/transports/postgres_arrow.rs b/connectorx/src/transports/postgres_arrow.rs
@@ -46,7 +46,7 @@ macro_rules! impl_postgres_transport {
             mappings = {
                 { Float4[f32]                        => Float32[f32]                           | conversion auto   }
                 { Float8[f64]                        => Float64[f64]                           | conversion auto   }
-                { Numeric[Decimal]                   => Float64[f64]                           | conversion option }
+                { Numeric[Decimal]                   => Decimal[Decimal]                       | conversion auto   }
                 { Int2[i16]                          => Int16[i16]                             | conversion auto   }
                 { Int4[i32]                          => Int32[i32]                             | conversion auto   }
                 { Int8[i64]                          => Int64[i64]                             | conversion auto   }
@@ -73,7 +73,7 @@ macro_rules! impl_postgres_transport {
                 { Int8Array[Vec<Option<i64>>]        => Int64Array[Vec<Option<i64>>]           | conversion auto   }
                 { Float4Array[Vec<Option<f32>>]      => Float32Array[Vec<Option<f32>>]         | conversion auto   }
                 { Float8Array[Vec<Option<f64>>]      => Float64Array[Vec<Option<f64>>]         | conversion auto   }
-                { NumericArray[Vec<Option<Decimal>>] => Float64Array[Vec<Option<f64>>]         | conversion option }
+                { NumericArray[Vec<Option<Decimal>>] => DecimalArray[Vec<Option<Decimal>>]     | conversion auto   }
             }
         );
     }
@@ -125,17 +125,4 @@ impl<P, C> TypeConversion<Value, String> for PostgresArrowTransport<P, C> {
     fn convert(val: Value) -> String {
         val.to_string()
     }
-}
-
-impl<P, C> TypeConversion<Vec<Option<Decimal>>, Vec<Option<f64>>> for PostgresArrowTransport<P, C> {
-    fn convert(val: Vec<Option<Decimal>>) -> Vec<Option<f64>> {
-        val.into_iter()
-            .map(|v| {
-                v.map(|v| {
-                    v.to_f64()
-                        .unwrap_or_else(|| panic!("cannot convert decimal {:?} to float64", v))
-                })
-            })
-            .collect()
-    }
-}
+}
diff --git a/connectorx/src/transports/postgres_arrowstream.rs b/connectorx/src/transports/postgres_arrowstream.rs
@@ -9,7 +9,6 @@ use crate::sources::postgres::{
 };
 use crate::typesystem::TypeConversion;
 use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime, Utc};
-use num_traits::ToPrimitive;
 use postgres::NoTls;
 use postgres_openssl::MakeTlsConnector;
 use rust_decimal::Decimal;
@@ -43,7 +42,7 @@ macro_rules! impl_postgres_transport {
             mappings = {
                 { Float4[f32]                => Float64[f64]              | conversion auto }
                 { Float8[f64]                => Float64[f64]              | conversion auto }
-                { Numeric[Decimal]           => Float64[f64]              | conversion option }
+                { Numeric[Decimal]           => Decimal[Decimal]          | conversion option }
                 { Int2[i16]                  => Int64[i64]                | conversion auto }
                 { Int4[i32]                  => Int64[i64]                | conversion auto }
                 { Int8[i64]                  => Int64[i64]                | conversion auto }
@@ -81,10 +80,9 @@ impl<P, C> TypeConversion<Uuid, String> for PostgresArrowTransport<P, C> {
     }
 }
 
-impl<P, C> TypeConversion<Decimal, f64> for PostgresArrowTransport<P, C> {
-    fn convert(val: Decimal) -> f64 {
-        val.to_f64()
-            .unwrap_or_else(|| panic!("cannot convert decimal {:?} to float64", val))
+impl<P, C> TypeConversion<Decimal, Decimal> for PostgresArrowTransport<P, C> {
+    fn convert(val: Decimal) -> Decimal {
+        val
     }
 }
 

diff --git a/connectorx/src/utils.rs b/connectorx/src/utils.rs
@@ -1,3 +1,5 @@
+use anyhow::Result;
+use rust_decimal::Decimal;
 use std::ops::{Deref, DerefMut};
 
 pub struct DummyBox<T>(pub T);
@@ -15,3 +17,18 @@ impl<T> DerefMut for DummyBox<T> {
         &mut self.0
     }
 }
+
+pub fn decimal_to_i128(mut v: Decimal, scale: u32) -> Result<i128> {
+    v.rescale(scale);
+
+    let v_scale = v.scale();
+    if v_scale != scale as u32 {
+        return Err(anyhow::anyhow!(
+            "decimal scale is not equal to expected scale, got: {} expected: {}",
+            v_scale,
+            scale
+        ));
+    }
+
+    Ok(v.mantissa())
+}