Merge pull request #869 from dbascoules/feat/oracle-decimal128

Support Oracle Decimal in arrowstream

Author: wangxiaoying

connectorx/Cargo.toml

@@ -84,7 +84,7 @@ src_csv = ["csv", "regex"]
 src_dummy = ["num-traits"]
 src_mssql = ["rust_decimal", "num-traits", "tiberius", "bb8-tiberius", "bb8", "tokio", "tokio-util", "uuid_old", "futures", "urlencoding"]
 src_mysql = ["r2d2_mysql", "mysql_common", "rust_decimal", "num-traits", "r2d2"]
-src_oracle = ["oracle", "r2d2-oracle","r2d2", "urlencoding"]
+src_oracle = ["oracle", "r2d2-oracle","r2d2", "urlencoding", "rust_decimal"]
 src_postgres = [
   "postgres",
   "r2d2_postgres",

connectorx/src/sources/oracle/errors.rs

@@ -18,6 +18,9 @@ pub enum OracleSourceError {
     #[error(transparent)]
     OracleUrlDecodeError(#[from] FromUtf8Error),
 
+    #[error(transparent)]
+    DecimalError(#[from] rust_decimal::Error),
+
     /// Any other errors that are too trivial to be put here explicitly.
     #[error(transparent)]
     Other(#[from] anyhow::Error),

connectorx/src/sources/oracle/mod.rs

@@ -23,6 +23,7 @@ use r2d2_oracle::{
     oracle::{Connector, Row, Statement},
     OracleConnectionManager,
 };
+use rust_decimal::Decimal;
 use sqlparser::dialect::Dialect;
 use url::Url;
 use urlencoding::decode;
@@ -372,3 +373,30 @@ impl_produce_text!(
     DateTime<Utc>,
     Vec<u8>,
 );
+
+// Manual implementation for Decimal since Oracle doesn't support it directly via FromSql
+impl<'r, 'a> Produce<'r, Decimal> for OracleTextSourceParser<'a> {
+    type Error = OracleSourceError;
+
+    #[throws(OracleSourceError)]
+    fn produce(&'r mut self) -> Decimal {
+        let (ridx, cidx) = self.next_loc()?;
+        let s: String = self.rowbuf[ridx].get(cidx)?;
+        let res = s.parse::<Decimal>()?;
+        res
+    }
+}
+
+impl<'r, 'a> Produce<'r, Option<Decimal>> for OracleTextSourceParser<'a> {
+    type Error = OracleSourceError;
+
+    #[throws(OracleSourceError)]
+    fn produce(&'r mut self) -> Option<Decimal> {
+        let (ridx, cidx) = self.next_loc()?;
+        let s: Option<String> = self.rowbuf[ridx].get(cidx)?;
+        match s {
+            Some(val) => Some(val.parse::<Decimal>()?),
+            None => None,
+        }
+    }
+}

connectorx/src/sources/oracle/typesystem.rs

@@ -1,9 +1,11 @@
 use chrono::{DateTime, NaiveDateTime, Utc};
 use r2d2_oracle::oracle::sql_type::OracleType;
+use rust_decimal::Decimal;
 
 #[derive(Copy, Clone, Debug)]
 pub enum OracleTypeSystem {
     NumInt(bool),
+    NumDecimal(bool),
     Float(bool),
     NumFloat(bool),
     BinaryFloat(bool),
@@ -25,6 +27,7 @@ impl_typesystem! {
     system = OracleTypeSystem,
     mappings = {
         { NumInt => i64 }
+        { NumDecimal => Decimal }
         { Float | NumFloat | BinaryFloat | BinaryDouble => f64 }
         { Blob => Vec<u8>}
         { Clob | VarChar | Char | NVarChar | NChar => String }
@@ -39,7 +42,7 @@ impl<'a> From<&'a OracleType> for OracleTypeSystem {
         match ty {
             OracleType::Number(0, 0) => NumFloat(true),
             OracleType::Number(_, 0) => NumInt(true),
-            OracleType::Number(_, _) => NumFloat(true),
+            OracleType::Number(_, _) => NumDecimal(true),
             OracleType::Float(_) => Float(true),
             OracleType::BinaryFloat => BinaryFloat(true),
             OracleType::BinaryDouble => BinaryDouble(true),

connectorx/src/transports/oracle_arrowstream.rs

@@ -7,6 +7,7 @@ use crate::{
     typesystem::TypeConversion,
 };
 use chrono::{DateTime, NaiveDateTime, Utc};
+use rust_decimal::Decimal;
 use thiserror::Error;
 
 #[derive(Error, Debug)]
@@ -30,6 +31,7 @@ impl_transport!(
     route = OracleSource => ArrowDestination,
     mappings = {
         { NumFloat[f64]              => Float64[f64]               | conversion auto }
+        { NumDecimal[Decimal]        => Decimal[Decimal]           | conversion auto }
         { Float[f64]                 => Float64[f64]               | conversion none }
         { BinaryFloat[f64]           => Float64[f64]               | conversion none }
         { BinaryDouble[f64]          => Float64[f64]               | conversion none }