feat(query): support desc table with decimal type and support order by decmail type col

src/query/expression/src/schema.rs

@@ -1330,14 +1330,12 @@ impl From<&DataType> for ArrowDataType {
             DataType::Number(ty) => with_number_type!(|TYPE| match ty {
                 NumberDataType::TYPE => ArrowDataType::TYPE,
             }),
-            DataType::Decimal(ty) => match ty {
-                DecimalDataType::Decimal128(s) => {
-                    ArrowDataType::Decimal(s.precision.into(), s.scale.into())
-                }
-                DecimalDataType::Decimal256(s) => {
-                    ArrowDataType::Decimal256(s.precision.into(), s.scale.into())
-                }
-            },
+            DataType::Decimal(DecimalDataType::Decimal128(s)) => {
+                ArrowDataType::Decimal(s.precision.into(), s.scale.into())
+            }
+            DataType::Decimal(DecimalDataType::Decimal256(s)) => {
+                ArrowDataType::Decimal256(s.precision.into(), s.scale.into())
+            }
             DataType::Timestamp => ArrowDataType::Timestamp(TimeUnit::Microsecond, None),
             DataType::Date => ArrowDataType::Date32,
             DataType::Nullable(ty) => ty.as_ref().into(),

src/query/expression/src/types.rs

@@ -298,6 +298,7 @@ impl DataType {
                 Scalar::Tuple(tys.iter().map(|ty| ty.default_value()).collect())
             }
             DataType::Variant => Scalar::Variant(vec![]),
+
             _ => unimplemented!(),
         }
     }

src/query/expression/src/types/decimal.rs

@@ -27,6 +27,9 @@ use serde::Deserialize;
 use serde::Serialize;
 
 use super::SimpleDomain;
+use crate::types::ArgType;
+use crate::types::DataType;
+use crate::types::GenericMap;
 use crate::types::ValueType;
 use crate::utils::arrow::buffer_into_mut;
 use crate::Column;
@@ -38,6 +41,9 @@ use crate::ScalarRef;
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct DecimalType<T: Decimal>(PhantomData<T>);
 
+pub type Decimal128Type = DecimalType<i128>;
+pub type Decimal256Type = DecimalType<i256>;
+
 impl<Num: Decimal> ValueType for DecimalType<Num> {
     type Scalar = Num;
     type ScalarRef<'a> = Num;
@@ -146,6 +152,38 @@ impl<Num: Decimal> ValueType for DecimalType<Num> {
     }
 }
 
+impl<Num: Decimal> ArgType for DecimalType<Num> {
+    fn data_type() -> DataType {
+        Num::data_type()
+    }
+
+    fn full_domain() -> Self::Domain {
+        SimpleDomain {
+            min: Num::MIN,
+            max: Num::MAX,
+        }
+    }
+
+    fn create_builder(capacity: usize, _generics: &GenericMap) -> Self::ColumnBuilder {
+        Vec::with_capacity(capacity)
+    }
+
+    fn column_from_vec(vec: Vec<Self::Scalar>, _generics: &GenericMap) -> Self::Column {
+        vec.into()
+    }
+
+    fn column_from_iter(iter: impl Iterator<Item = Self::Scalar>, _: &GenericMap) -> Self::Column {
+        iter.collect()
+    }
+
+    fn column_from_ref_iter<'a>(
+        iter: impl Iterator<Item = Self::ScalarRef<'a>>,
+        _: &GenericMap,
+    ) -> Self::Column {
+        iter.collect()
+    }
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, EnumAsInner)]
 pub enum DecimalDataType {
     Decimal128(DecimalSize),
@@ -228,6 +266,9 @@ pub trait Decimal:
     fn upcast_scalar(scalar: Self) -> Scalar;
     fn upcast_column(col: Buffer<Self>) -> Column;
     fn upcast_domain(domain: SimpleDomain<Self>) -> Domain;
+    fn data_type() -> DataType;
+    const MIN: Self;
+    const MAX: Self;
 
     fn to_column_from_buffer(value: Buffer<Self>, size: DecimalSize) -> DecimalColumn;
 
@@ -262,14 +303,14 @@ impl Decimal for i128 {
         -1_i128
     }
 
-    fn mem_size() -> usize {
-        16
-    }
-
     fn e(n: u32) -> Self {
         10_i128.pow(n)
     }
 
+    fn mem_size() -> usize {
+        16
+    }
+
     fn checked_add(self, rhs: Self) -> Option<Self> {
         self.checked_add(rhs)
     }
@@ -363,6 +404,16 @@ impl Decimal for i128 {
         }))
     }
 
+    fn data_type() -> DataType {
+        DataType::Decimal(DecimalDataType::Decimal128(DecimalSize {
+            precision: MAX_DECIMAL128_PRECISION,
+            scale: 0,
+        }))
+    }
+
+    const MIN: i128 = i128::MIN;
+
+    const MAX: i128 = i128::MAX;
     fn to_column_from_buffer(value: Buffer<Self>, size: DecimalSize) -> DecimalColumn {
         DecimalColumn::Decimal128(value, size)
     }
@@ -481,6 +532,15 @@ impl Decimal for i256 {
         }))
     }
 
+    fn data_type() -> DataType {
+        DataType::Decimal(DecimalDataType::Decimal128(DecimalSize {
+            precision: MAX_DECIMAL256_PRECISION,
+            scale: 0,
+        }))
+    }
+
+    const MIN: i256 = i256::MIN;
+    const MAX: i256 = i256::MAX;
     fn to_column_from_buffer(value: Buffer<Self>, size: DecimalSize) -> DecimalColumn {
         DecimalColumn::Decimal256(value, size)
     }

src/query/expression/src/utils/column_from.rs

@@ -14,6 +14,9 @@
 
 use std::iter::Iterator;
 
+use ethnum::i256;
+
+use crate::types::decimal::*;
 use crate::types::nullable::NullableColumn;
 use crate::types::number::*;
 use crate::types::*;
@@ -129,6 +132,24 @@ impl<D: AsRef<[f64]>> FromData<D, [Vec<f64>; 0]> for Float64Type {
     }
 }
 
+impl<D: AsRef<[i128]>> FromData<D, [Vec<i128>; 0]> for Decimal128Type {
+    fn from_data(d: D) -> Column {
+        Decimal128Type::upcast_column(Decimal128Type::column_from_iter(
+            d.as_ref().iter().copied(),
+            &[],
+        ))
+    }
+}
+
+impl<D: AsRef<[i256]>> FromData<D, [Vec<i256>; 0]> for Decimal256Type {
+    fn from_data(d: D) -> Column {
+        Decimal256Type::upcast_column(Decimal256Type::column_from_iter(
+            d.as_ref().iter().copied(),
+            &[],
+        ))
+    }
+}
+
 #[cfg(test)]
 mod test {
 

src/query/expression/src/utils/display.rs

@@ -781,7 +781,7 @@ fn display_decimal_256(num: i256, scale: u8) -> String {
                 buf,
                 "{}.{:0>width$}",
                 num / pow_scale,
-                num % pow_scale,
+                num % pow_scale.abs(),
                 width = scale as usize
             )
             .unwrap();

src/query/expression/src/values.rs

@@ -489,6 +489,7 @@ impl PartialOrd for Column {
                 col1.partial_cmp(col2)
             }
             (Column::Number(col1), Column::Number(col2)) => col1.partial_cmp(col2),
+            (Column::Decimal(col1), Column::Decimal(col2)) => col1.partial_cmp(col2),
             (Column::Boolean(col1), Column::Boolean(col2)) => col1.iter().partial_cmp(col2.iter()),
             (Column::String(col1), Column::String(col2)) => col1.iter().partial_cmp(col2.iter()),
             (Column::Timestamp(col1), Column::Timestamp(col2)) => {
@@ -1502,6 +1503,7 @@ impl ColumnBuilder {
             (ColumnBuilder::Null { len }, ScalarRef::Null) => *len += 1,
             (ColumnBuilder::EmptyArray { len }, ScalarRef::EmptyArray) => *len += 1,
             (ColumnBuilder::Number(builder), ScalarRef::Number(value)) => builder.push(value),
+            (ColumnBuilder::Decimal(builder), ScalarRef::Decimal(value)) => builder.push(value),
             (ColumnBuilder::Boolean(builder), ScalarRef::Boolean(value)) => builder.push(value),
             (ColumnBuilder::String(builder), ScalarRef::String(value)) => {
                 builder.put_slice(value);
@@ -1571,6 +1573,9 @@ impl ColumnBuilder {
             (ColumnBuilder::Number(builder), Column::Number(column)) => {
                 builder.append_column(column);
             }
+            (ColumnBuilder::Decimal(builder), Column::Decimal(column)) => {
+                builder.append_column(column);
+            }
             (ColumnBuilder::Boolean(builder), Column::Boolean(other)) => {
                 append_bitmap(builder, other);
             }

src/query/expression/tests/it/sort.rs

@@ -16,6 +16,7 @@ use std::sync::Arc;
 use std::vec;
 
 use common_exception::Result;
+use common_expression::types::decimal::*;
 use common_expression::types::number::*;
 use common_expression::types::StringType;
 use common_expression::Column;
@@ -108,6 +109,87 @@ fn test_block_sort() -> Result<()> {
         }
     }
 
+    let decimal_block = new_block(&[
+        Decimal128Type::from_data(vec![6i128, 4, 3, 2, 1, 1, 7]),
+        StringType::from_data(vec!["b1", "b2", "b3", "b4", "b5", "b6", "b7"]),
+    ]);
+
+    // test cast:
+    // - sort descriptions
+    // - limit
+    // - expected cols
+    let test_cases: Vec<(Vec<SortColumnDescription>, Option<usize>, Vec<Column>)> = vec![
+        (
+            vec![SortColumnDescription {
+                offset: 0,
+                asc: true,
+                nulls_first: false,
+            }],
+            None,
+            vec![
+                Decimal128Type::from_data(vec![1_i128, 1, 2, 3, 4, 6, 7]),
+                StringType::from_data(vec!["b5", "b6", "b4", "b3", "b2", "b1", "b7"]),
+            ],
+        ),
+        (
+            vec![SortColumnDescription {
+                offset: 0,
+                asc: true,
+                nulls_first: false,
+            }],
+            Some(4),
+            vec![
+                Decimal128Type::from_data(vec![1_i128, 1, 2, 3]),
+                StringType::from_data(vec!["b5", "b6", "b4", "b3"]),
+            ],
+        ),
+        (
+            vec![SortColumnDescription {
+                offset: 1,
+                asc: false,
+                nulls_first: false,
+            }],
+            None,
+            vec![
+                Decimal128Type::from_data(vec![7_i128, 1, 1, 2, 3, 4, 6]),
+                StringType::from_data(vec!["b7", "b6", "b5", "b4", "b3", "b2", "b1"]),
+            ],
+        ),
+        (
+            vec![
+                SortColumnDescription {
+                    offset: 0,
+                    asc: true,
+                    nulls_first: false,
+                },
+                SortColumnDescription {
+                    offset: 1,
+                    asc: false,
+                    nulls_first: false,
+                },
+            ],
+            None,
+            vec![
+                Decimal128Type::from_data(vec![1_i128, 1, 2, 3, 4, 6, 7]),
+                StringType::from_data(vec!["b6", "b5", "b4", "b3", "b2", "b1", "b7"]),
+            ],
+        ),
+    ];
+
+    for (sort_descs, limit, expected) in test_cases {
+        let res = DataBlock::sort(&decimal_block, &sort_descs, limit)?;
+
+        for (entry, expect) in res.columns().iter().zip(expected.iter()) {
+            assert_eq!(
+                entry.value.as_column().unwrap(),
+                expect,
+                "the column after sort is wrong, expect: {:?}, got: {:?}",
+                expect,
+                entry.value
+            );
+        }
+    }
+
     Ok(())
 }
 

src/query/storages/common/index/src/index.rs

@@ -20,7 +20,11 @@ pub trait Index {
         let inner_type = data_type.remove_nullable();
         matches!(
             inner_type,
-            DataType::Number(_) | DataType::Date | DataType::Timestamp | DataType::String
+            DataType::Number(_)
+                | DataType::Date
+                | DataType::Timestamp
+                | DataType::String
+                | DataType::Decimal(_)
         )
     }
 }