feat(constants): introduce default Decimal128 constants for Arrow integration

- Added DEFAULT_ARROW_DECIMAL_PRECISION, DEFAULT_ARROW_DECIMAL_SCALE, and DEFAULT_ARROW_DECIMAL constants to support Decimal128 in Arrow.
- Updated imports in arrow_assoc modules to utilize the new constants.
- Removed redundant constant definitions from arrow_assoc implementations.

Author: longshan.lu

connectorx/src/constants.rs

@@ -1,3 +1,16 @@
+#[cfg(feature = "dst_arrow")]
+use arrow::datatypes::DataType as ArrowDataType;
+
+#[cfg(feature = "dst_arrow")]
+pub const DEFAULT_ARROW_DECIMAL_PRECISION: u8 = 38;
+
+#[cfg(feature = "dst_arrow")]
+pub const DEFAULT_ARROW_DECIMAL_SCALE: i8 = 10;
+
+#[cfg(feature = "dst_arrow")]
+pub const DEFAULT_ARROW_DECIMAL: ArrowDataType =
+    ArrowDataType::Decimal128(DEFAULT_ARROW_DECIMAL_PRECISION, DEFAULT_ARROW_DECIMAL_SCALE);
+
 #[cfg(feature = "dst_arrow")]
 pub(crate) const SECONDS_IN_DAY: i64 = 86_400;
 

connectorx/src/destinations/arrow/arrow_assoc.rs

@@ -2,7 +2,7 @@ use super::{
     errors::{ArrowDestinationError, Result},
     typesystem::{DateTimeWrapperMicro, NaiveDateTimeWrapperMicro, NaiveTimeWrapperMicro},
 };
-use crate::{constants::SECONDS_IN_DAY, utils::decimal_to_i128};
+use crate::{constants::{DEFAULT_ARROW_DECIMAL, DEFAULT_ARROW_DECIMAL_SCALE, SECONDS_IN_DAY}, utils::decimal_to_i128};
 use arrow::array::{
     ArrayBuilder, BooleanBuilder, Date32Builder, Decimal128Builder, Float32Builder, Float64Builder,
     Int16Builder, Int32Builder, Int64Builder, LargeBinaryBuilder, LargeListBuilder, StringBuilder,
@@ -72,10 +72,7 @@ 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;

connectorx/src/destinations/arrowstream/arrow_assoc.rs

@@ -1,5 +1,5 @@
 use super::errors::{ArrowDestinationError, Result};
-use crate::constants::SECONDS_IN_DAY;
+use crate::constants::{DEFAULT_ARROW_DECIMAL, DEFAULT_ARROW_DECIMAL_SCALE, SECONDS_IN_DAY};
 use crate::utils::decimal_to_i128;
 use arrow::array::{
     ArrayBuilder, BooleanBuilder, Date32Builder, Date64Builder, Decimal128Builder, Float32Builder,
@@ -11,11 +11,6 @@ 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;

connectorx/src/transports/postgres_arrowstream.rs

@@ -42,7 +42,7 @@ macro_rules! impl_postgres_transport {
             mappings = {
                 { Float4[f32]                => Float64[f64]              | conversion auto }
                 { Float8[f64]                => Float64[f64]              | conversion auto }
-                { Numeric[Decimal]           => Decimal[Decimal]          | conversion option }
+                { Numeric[Decimal]           => Decimal[Decimal]          | conversion auto }
                 { Int2[i16]                  => Int64[i64]                | conversion auto }
                 { Int4[i32]                  => Int64[i64]                | conversion auto }
                 { Int8[i64]                  => Int64[i64]                | conversion auto }
@@ -80,12 +80,6 @@ impl<P, C> TypeConversion<Uuid, String> for PostgresArrowTransport<P, C> {
     }
 }
 
-impl<P, C> TypeConversion<Decimal, Decimal> for PostgresArrowTransport<P, C> {
-    fn convert(val: Decimal) -> Decimal {
-        val
-    }
-}
-
 impl<P, C> TypeConversion<Value, String> for PostgresArrowTransport<P, C> {
     fn convert(val: Value) -> String {
         val.to_string()

connectorx/tests/test_postgres.rs

@@ -26,46 +26,6 @@ use postgres::NoTls;
 use std::env;
 use url::Url;
 
-#[test]
-fn test_types_simple_postgres_aa() {
-    let _ = env_logger::builder().is_test(true).try_init();
-
-    let dburl = env::var("POSTGRES_URL").unwrap();
-
-    let vars = vec!["test_numeric"].join(",");
-
-    let queries = [CXQuery::naked(format!("select {vars} from test_types"))];
-    let url = Url::parse(dburl.as_str()).unwrap();
-    let (config, _tls) = rewrite_tls_args(&url).unwrap();
-    let builder = PostgresSource::<SimpleProtocol, NoTls>::new(config, NoTls, 2).unwrap();
-    let mut destination = ArrowDestination::new();
-    let dispatcher = Dispatcher::<_, _, PostgresArrowTransport<SimpleProtocol, NoTls>>::new(
-        builder,
-        &mut destination,
-        &queries,
-        Some(String::from("select * from test_types")),
-    );
-
-    dispatcher.run().expect("run dispatcher");
-
-    let result = destination.arrow().unwrap();
-
-    arrow::util::pretty::print_batches(&result).unwrap();
-}
-
-#[test]
-fn test_decimal_128() {
-    let mut builder = Decimal128Builder::new();
-    builder.append_value(1);
-    builder.append_value(1234567890);
-    builder.append_value(1234567890);
-    builder.append_value(1234567890);
-    builder.append_value(1234567890);
-    let decimal = builder.finish();
-
-    println!("decimal: {:#?}", decimal.value_as_string(0));
-}
-
 #[test]
 fn load_and_parse() {
     let _ = env_logger::builder().is_test(true).try_init();