chore: decimal

Cargo.toml

@@ -134,6 +134,7 @@ common-base = { path = "src/common/base" }
 common-catalog = { path = "src/common/catalog" }
 common-config = { path = "src/common/config" }
 common-datasource = { path = "src/common/datasource" }
+common-decimal = { path = "src/common/decimal" }
 common-error = { path = "src/common/error" }
 common-function = { path = "src/common/function" }
 common-greptimedb-telemetry = { path = "src/common/greptimedb-telemetry" }

src/api/src/helper.rs

@@ -158,7 +158,8 @@ impl TryFrom<ConcreteDataType> for ColumnDataTypeWrapper {
             },
             ConcreteDataType::Null(_)
             | ConcreteDataType::List(_)
-            | ConcreteDataType::Dictionary(_) => {
+            | ConcreteDataType::Dictionary(_)
+            | ConcreteDataType::Decimal128(_) => {
                 return error::IntoColumnDataTypeSnafu { from: datatype }.fail()
             }
         });
@@ -341,7 +342,7 @@ pub fn push_vals(column: &mut Column, origin_count: usize, vector: VectorRef) {
             TimeUnit::Microsecond => values.duration_microsecond_values.push(val.value()),
             TimeUnit::Nanosecond => values.duration_nanosecond_values.push(val.value()),
         },
-        Value::List(_) => unreachable!(),
+        Value::List(_) | Value::Decimal128(_) => unreachable!(),
     });
     column.null_mask = null_mask.into_vec();
 }
@@ -522,7 +523,10 @@ pub fn pb_values_to_vector_ref(data_type: &ConcreteDataType, values: Values) ->
                 values.duration_nanosecond_values,
             )),
         },
-        ConcreteDataType::Null(_) | ConcreteDataType::List(_) | ConcreteDataType::Dictionary(_) => {
+        ConcreteDataType::Null(_)
+        | ConcreteDataType::List(_)
+        | ConcreteDataType::Dictionary(_)
+        | ConcreteDataType::Decimal128(_) => {
             unreachable!()
         }
     }
@@ -692,7 +696,10 @@ pub fn pb_values_to_values(data_type: &ConcreteDataType, values: Values) -> Vec<
             .into_iter()
             .map(|v| Value::Duration(Duration::new_nanosecond(v)))
             .collect(),
-        ConcreteDataType::Null(_) | ConcreteDataType::List(_) | ConcreteDataType::Dictionary(_) => {
+        ConcreteDataType::Null(_)
+        | ConcreteDataType::List(_)
+        | ConcreteDataType::Dictionary(_)
+        | ConcreteDataType::Decimal128(_) => {
             unreachable!()
         }
     }
@@ -816,7 +823,7 @@ pub fn to_proto_value(value: Value) -> Option<v1::Value> {
                 value_data: Some(ValueData::DurationNanosecondValue(v.value())),
             },
         },
-        Value::List(_) => return None,
+        Value::List(_) | Value::Decimal128(_) => return None,
     };
 
     Some(proto_value)
@@ -908,9 +915,10 @@ pub fn to_column_data_type(data_type: &ConcreteDataType) -> Option<ColumnDataTyp
             ColumnDataType::IntervalMonthDayNano
         }
         ConcreteDataType::Interval(IntervalType::DayTime(_)) => ColumnDataType::IntervalDayTime,
-        ConcreteDataType::Null(_) | ConcreteDataType::List(_) | ConcreteDataType::Dictionary(_) => {
-            return None
-        }
+        ConcreteDataType::Null(_)
+        | ConcreteDataType::List(_)
+        | ConcreteDataType::Dictionary(_)
+        | ConcreteDataType::Decimal128(_) => return None,
     };
 
     Some(column_data_type)
@@ -974,7 +982,7 @@ pub fn value_to_grpc_value(value: Value) -> GrpcValue {
                 TimeUnit::Microsecond => ValueData::DurationMicrosecondValue(v.value()),
                 TimeUnit::Nanosecond => ValueData::DurationNanosecondValue(v.value()),
             }),
-            Value::List(_) => unreachable!(),
+            Value::List(_) | Value::Decimal128(_) => unreachable!(),
         },
     }
 }

src/common/decimal/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "decimal"
+name = "common-decimal"
 version.workspace = true
 edition.workspace = true
 license.workspace = true

src/common/decimal/src/decimal128.rs

@@ -43,7 +43,7 @@ const BYTES_TO_OVERFLOW_RUST_DECIMAL: usize = 28;
 /// **precision**: the total number of digits in the number, it's range is \[1, 38\].
 ///
 /// **scale**: the number of digits to the right of the decimal point, it's range is \[0, precision\].
-#[derive(Debug, Default, Eq, Copy, Clone, Serialize, Deserialize)]
+#[derive(Debug, Eq, Copy, Clone, Serialize, Deserialize)]
 pub struct Decimal128 {
     value: i128,
     precision: u8,
@@ -90,6 +90,17 @@ impl Decimal128 {
     }
 }
 
+/// The default value of Decimal128 is 0, and its precision is 1 and scale is 0.
+impl Default for Decimal128 {
+    fn default() -> Self {
+        Self {
+            value: 0,
+            precision: 1,
+            scale: 0,
+        }
+    }
+}
+
 impl PartialEq for Decimal128 {
     fn eq(&self, other: &Self) -> bool {
         self.precision.eq(&other.precision)

src/common/decimal/src/lib.rs

@@ -14,3 +14,5 @@
 
 pub mod decimal128;
 pub mod error;
+
+pub use decimal128::Decimal128;

src/common/grpc/src/select.rs

@@ -71,7 +71,7 @@ macro_rules! convert_arrow_array_to_grpc_vals {
                     return Ok(vals);
                 },
             )+
-            ConcreteDataType::Null(_) | ConcreteDataType::List(_) | ConcreteDataType::Dictionary(_) => unreachable!("Should not send {:?} in gRPC", $data_type),
+            ConcreteDataType::Null(_) | ConcreteDataType::List(_) | ConcreteDataType::Dictionary(_) | ConcreteDataType::Decimal128(_) => unreachable!("Should not send {:?} in gRPC", $data_type),
         }
     }};
 }

src/datatypes/Cargo.toml

@@ -13,6 +13,7 @@ arrow-array.workspace = true
 arrow-schema.workspace = true
 arrow.workspace = true
 common-base.workspace = true
+common-decimal.workspace = true
 common-error.workspace = true
 common-macro.workspace = true
 common-telemetry.workspace = true
@@ -23,6 +24,7 @@ num = "0.4"
 num-traits = "0.2"
 ordered-float = { version = "3.0", features = ["serde"] }
 paste = "1.0"
+rust_decimal = "1.32.0"
 serde.workspace = true
 serde_json = "1.0"
 snafu.workspace = true

src/datatypes/src/data_type.rs

@@ -19,6 +19,8 @@ use arrow::compute::cast as arrow_array_cast;
 use arrow::datatypes::{
     DataType as ArrowDataType, IntervalUnit as ArrowIntervalUnit, TimeUnit as ArrowTimeUnit,
 };
+use arrow_schema::DECIMAL_DEFAULT_SCALE;
+use common_decimal::decimal128::DECIMAL128_MAX_PRECISION;
 use common_time::interval::IntervalUnit;
 use common_time::timestamp::TimeUnit;
 use paste::paste;
@@ -246,6 +248,13 @@ impl ConcreteDataType {
         }
     }
 
+    pub fn as_decimal(&self) -> Option<DecimalType> {
+        match self {
+            ConcreteDataType::Decimal128(d) => Some(*d),
+            _ => None,
+        }
+    }
+
     /// Checks if the data type can cast to another data type.
     pub fn can_arrow_type_cast_to(&self, to_type: &ConcreteDataType) -> bool {
         let array = arrow_array::new_empty_array(&self.as_arrow_type());
@@ -472,6 +481,10 @@ impl ConcreteDataType {
     pub fn decimal128_datatype(precision: u8, scale: i8) -> ConcreteDataType {
         ConcreteDataType::Decimal128(DecimalType::new(precision, scale))
     }
+
+    pub fn decimal128_default_datatype() -> ConcreteDataType {
+        Self::decimal128_datatype(DECIMAL128_MAX_PRECISION, DECIMAL_DEFAULT_SCALE)
+    }
 }
 
 /// Data type abstraction.

src/datatypes/src/error.rs

@@ -122,6 +122,13 @@ pub enum Error {
 
     #[snafu(display("Failed to unpack value to given type: {}", reason))]
     TryFromValue { reason: String, location: Location },
+
+    #[snafu(display("Invalid arguments, reason: {}", error))]
+    InvalidArguments {
+        #[snafu(source)]
+        error: arrow::error::ArrowError,
+        location: Location,
+    },
 }
 
 impl ErrorExt for Error {

src/datatypes/src/scalars.rs

@@ -14,6 +14,7 @@
 
 use std::any::Any;
 
+use common_decimal::Decimal128;
 use common_time::{Date, DateTime};
 
 use crate::types::{
@@ -22,8 +23,8 @@ use crate::types::{
 };
 use crate::value::{ListValue, ListValueRef, Value};
 use crate::vectors::{
-    BinaryVector, BooleanVector, DateTimeVector, DateVector, ListVector, MutableVector,
-    PrimitiveVector, StringVector, Vector,
+    BinaryVector, BooleanVector, DateTimeVector, DateVector, Decimal128Vector, ListVector,
+    MutableVector, PrimitiveVector, StringVector, Vector,
 };
 
 fn get_iter_capacity<T, I: Iterator<Item = T>>(iter: &I) -> usize {
@@ -277,6 +278,27 @@ impl<'a> ScalarRef<'a> for Date {
     }
 }
 
+impl Scalar for Decimal128 {
+    type VectorType = Decimal128Vector;
+    type RefType<'a> = Decimal128;
+
+    fn as_scalar_ref(&self) -> Self::RefType<'_> {
+        *self
+    }
+
+    fn upcast_gat<'short, 'long: 'short>(long: Self::RefType<'long>) -> Self::RefType<'short> {
+        long
+    }
+}
+
+impl<'a> ScalarRef<'a> for Decimal128 {
+    type ScalarType = Decimal128;
+
+    fn to_owned_scalar(&self) -> Self::ScalarType {
+        *self
+    }
+}
+
 impl Scalar for DateTime {
     type VectorType = DateTimeVector;
     type RefType<'a> = DateTime;
@@ -396,6 +418,13 @@ mod tests {
         assert_eq!(date, date.to_owned_scalar());
     }
 
+    #[test]
+    fn test_decimal_scalar() {
+        let decimal = Decimal128::new_unchecked(1, 1, 1);
+        assert_eq!(decimal, decimal.as_scalar_ref());
+        assert_eq!(decimal, decimal.to_owned_scalar());
+    }
+
     #[test]
     fn test_datetime_scalar() {
         let dt = DateTime::new(123);

src/datatypes/src/type_id.rs

@@ -30,6 +30,8 @@ pub enum LogicalTypeId {
     Float32,
     Float64,
 
+    Decimal128,
+
     // String types:
     String,
     Binary,
@@ -123,6 +125,7 @@ impl LogicalTypeId {
             LogicalTypeId::DurationMillisecond => ConcreteDataType::duration_millisecond_datatype(),
             LogicalTypeId::DurationMicrosecond => ConcreteDataType::duration_microsecond_datatype(),
             LogicalTypeId::DurationNanosecond => ConcreteDataType::duration_nanosecond_datatype(),
+            LogicalTypeId::Decimal128 => ConcreteDataType::decimal128_default_datatype(),
         }
     }
 }

src/datatypes/src/types.rs

@@ -17,6 +17,7 @@ mod boolean_type;
 pub mod cast;
 mod date_type;
 mod datetime_type;
+mod decimal_type;
 mod dictionary_type;
 mod duration_type;
 mod interval_type;
@@ -29,9 +30,10 @@ mod timestamp_type;
 
 pub use binary_type::BinaryType;
 pub use boolean_type::BooleanType;
-pub use cast::cast_with_opt;
+pub use cast::{cast, cast_with_opt};
 pub use date_type::DateType;
 pub use datetime_type::DateTimeType;
+pub use decimal_type::DecimalType;
 pub use dictionary_type::DictionaryType;
 pub use duration_type::{
     DurationMicrosecondType, DurationMillisecondType, DurationNanosecondType, DurationSecondType,

src/datatypes/src/types/decimal_type.rs

@@ -22,7 +22,9 @@ use crate::value::Value;
 use crate::vectors::{Decimal128VectorBuilder, MutableVector};
 
 /// Decimal type with precision and scale information.
-#[derive(Debug, Default, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
+#[derive(
+    Debug, Default, Copy, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize,
+)]
 pub struct DecimalType {
     precision: u8,
     scale: i8,

src/datatypes/src/types/string_type.rs

@@ -80,6 +80,7 @@ impl DataType for StringType {
             Value::Time(v) => Some(Value::String(StringBytes::from(v.to_iso8601_string()))),
             Value::Interval(v) => Some(Value::String(StringBytes::from(v.to_iso8601_string()))),
             Value::Duration(v) => Some(Value::String(StringBytes::from(v.to_string()))),
+            Value::Decimal128(v) => Some(Value::String(StringBytes::from(v.to_string()))),
 
             // StringBytes is only support for utf-8, Value::Binary is not allowed.
             Value::Binary(_) | Value::List(_) => None,