chore(query): Simplified sort simple rows implementation

src/query/pipeline/transforms/src/processors/transforms/sort/cursor.rs

@@ -15,8 +15,6 @@
 use std::cmp::Ordering;
 use std::marker::PhantomData;
 
-use databend_common_expression::Column;
-
 use super::rows::Rows;
 
 /// A cursor point to a certain row in a data block.
@@ -72,11 +70,6 @@ where
         self.num_rows
     }
 
-    #[inline]
-    pub fn to_column(&self) -> Column {
-        self.rows.to_column()
-    }
-
     pub fn cursor_mut(&self) -> CursorMut<'_, R, O> {
         CursorMut {
             row_index: self.row_index,

src/query/pipeline/transforms/src/processors/transforms/sort/rows/common.rs

@@ -16,6 +16,7 @@ use databend_common_exception::Result;
 use databend_common_expression::types::binary::BinaryColumn;
 use databend_common_expression::types::binary::BinaryColumnBuilder;
 use databend_common_expression::types::nullable::NullableColumn;
+use databend_common_expression::types::BinaryType;
 use databend_common_expression::types::DataType;
 use databend_common_expression::BlockEntry;
 use databend_common_expression::Column;
@@ -35,6 +36,7 @@ pub type CommonRows = BinaryColumn;
 
 impl Rows for BinaryColumn {
     type Item<'a> = &'a [u8];
+    type Type = BinaryType;
 
     fn len(&self) -> usize {
         self.len()
@@ -51,10 +53,6 @@ impl Rows for BinaryColumn {
     fn try_from_column(col: &Column, _: &[SortColumnDescription]) -> Option<Self> {
         col.as_binary().cloned()
     }
-
-    fn data_type() -> DataType {
-        DataType::Binary
-    }
 }
 
 impl RowConverter<BinaryColumn> for CommonRowConverter {

src/query/pipeline/transforms/src/processors/transforms/sort/rows/mod.rs

@@ -18,6 +18,7 @@ mod simple;
 pub use common::*;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_expression::types::ArgType;
 use databend_common_expression::types::DataType;
 use databend_common_expression::BlockEntry;
 use databend_common_expression::Column;
@@ -42,6 +43,7 @@ where Self: Sized + Clone
 {
     type Item<'a>: Ord
     where Self: 'a;
+    type Type: ArgType;
 
     fn len(&self) -> usize;
     fn row(&self, index: usize) -> Self::Item<'_>;
@@ -56,9 +58,12 @@ where Self: Sized + Clone
             ))
         })
     }
+
     fn try_from_column(col: &Column, desc: &[SortColumnDescription]) -> Option<Self>;
 
-    fn data_type() -> DataType;
+    fn data_type() -> DataType {
+        Self::Type::data_type()
+    }
 
     fn is_empty(&self) -> bool {
         self.len() == 0

src/query/pipeline/transforms/src/processors/transforms/sort/rows/simple.rs

@@ -12,16 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::cmp::Ordering;
+use std::cmp::Reverse;
 use std::marker::PhantomData;
 
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::types::ArgType;
-use databend_common_expression::types::DataType;
-use databend_common_expression::types::DateType;
-use databend_common_expression::types::StringType;
-use databend_common_expression::types::TimestampType;
 use databend_common_expression::types::ValueType;
 use databend_common_expression::BlockEntry;
 use databend_common_expression::Column;
@@ -33,82 +29,68 @@ use databend_common_expression::Value;
 use super::RowConverter;
 use super::Rows;
 
-pub type DateRows = SimpleRows<DateType>;
-pub type TimestampRows = SimpleRows<TimestampType>;
-pub type StringRows = SimpleRows<StringType>;
-
-/// Row structure for single simple types. (numbers, date, timestamp)
-#[derive(Clone, Copy)]
-pub struct SimpleRow<T: ValueType> {
-    inner: T::Scalar,
-    desc: bool,
-}
-
 /// Rows structure for single simple types. (numbers, date, timestamp)
 #[derive(Clone)]
-pub struct SimpleRows<T: ValueType> {
+pub struct SimpleRowsAsc<T: ValueType> {
     inner: T::Column,
-    desc: bool,
 }
 
-impl<T> Ord for SimpleRow<T>
+impl<T> Rows for SimpleRowsAsc<T>
 where
-    T: ValueType,
-    T::Scalar: Ord,
+    T: ArgType,
+    for<'a> T::ScalarRef<'a>: Ord,
 {
-    fn cmp(&self, other: &Self) -> Ordering {
-        if self.desc {
-            self.inner.cmp(&other.inner).reverse()
-        } else {
-            self.inner.cmp(&other.inner)
-        }
+    type Item<'a> = T::ScalarRef<'a>
+    where Self: 'a;
+
+    type Type = T;
+
+    fn len(&self) -> usize {
+        T::column_len(&self.inner)
     }
-}
 
-impl<T> PartialOrd for SimpleRow<T>
-where
-    T: ValueType,
-    T::Scalar: Ord,
-{
-    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
-        Some(self.cmp(other))
+    fn row(&self, index: usize) -> T::ScalarRef<'_> {
+        unsafe { T::index_column_unchecked(&self.inner, index) }
     }
-}
 
-impl<T> PartialEq for SimpleRow<T>
-where
-    T: ValueType,
-    T::Scalar: Ord,
-{
-    fn eq(&self, other: &Self) -> bool {
-        self.inner == other.inner && self.desc == other.desc
+    fn to_column(&self) -> Column {
+        T::upcast_column(self.inner.clone())
+    }
+
+    fn try_from_column(col: &Column, desc: &[SortColumnDescription]) -> Option<Self> {
+        let inner = T::try_downcast_column(col)?;
+
+        if desc[0].asc {
+            Some(Self { inner })
+        } else {
+            None
+        }
     }
 }
 
-impl<T> Eq for SimpleRow<T>
-where
-    T: ValueType,
-    T::Scalar: Ord,
-{
+/// Rows structure for single simple types. (numbers, date, timestamp)
+#[derive(Clone)]
+pub struct SimpleRowsDesc<T: ValueType> {
+    inner: T::Column,
 }
 
-impl<T> Rows for SimpleRows<T>
+impl<T> Rows for SimpleRowsDesc<T>
 where
     T: ArgType,
-    T::Scalar: Ord,
+    for<'a> T::ScalarRef<'a>: Ord,
 {
-    type Item<'a> = SimpleRow<T>;
+    type Item<'a> = Reverse<T::ScalarRef<'a>>
+    where Self: 'a;
+
+    type Type = T;
 
     fn len(&self) -> usize {
         T::column_len(&self.inner)
     }
 
-    fn row(&self, index: usize) -> Self::Item<'_> {
-        let inner = unsafe { T::index_column_unchecked(&self.inner, index) };
-        SimpleRow {
-            inner: T::to_owned_scalar(inner),
-            desc: self.desc,
-        }
+    fn row(&self, index: usize) -> Reverse<T::ScalarRef<'_>> {
+        let value = unsafe { T::index_column_unchecked(&self.inner, index) };
+        Reverse(value)
     }
 
     fn to_column(&self) -> Column {
@@ -117,46 +99,75 @@ where
 
     fn try_from_column(col: &Column, desc: &[SortColumnDescription]) -> Option<Self> {
         let inner = T::try_downcast_column(col)?;
-        Some(Self {
-            inner,
-            desc: !desc[0].asc,
-        })
-    }
 
-    fn data_type() -> DataType {
-        T::data_type()
+        if !desc[0].asc {
+            Some(Self { inner })
+        } else {
+            None
+        }
     }
 }
 
-pub type DateConverter = SimpleRowConverter<DateType>;
-pub type TimestampConverter = SimpleRowConverter<TimestampType>;
-pub type StringConverter = SimpleRowConverter<StringType>;
-
 /// If there is only one sort field and its type is a primitive type,
 /// use this converter.
 pub struct SimpleRowConverter<T> {
-    desc: bool,
     _t: PhantomData<T>,
 }
 
-impl<T> RowConverter<SimpleRows<T>> for SimpleRowConverter<T>
+impl<T> RowConverter<SimpleRowsAsc<T>> for SimpleRowConverter<T>
 where
     T: ArgType,
-    T::Scalar: Ord,
+    for<'a> T::ScalarRef<'a>: Ord,
 {
     fn create(
         sort_columns_descriptions: &[SortColumnDescription],
         _: DataSchemaRef,
     ) -> Result<Self> {
         assert!(sort_columns_descriptions.len() == 1);
+        assert!(sort_columns_descriptions[0].asc);
+        Ok(Self { _t: PhantomData })
+    }
+
+    fn convert(&mut self, columns: &[BlockEntry], num_rows: usize) -> Result<SimpleRowsAsc<T>> {
+        assert!(columns.len() == 1);
+        let col = &columns[0];
+        if col.data_type != T::data_type() {
+            return Err(ErrorCode::Internal(format!(
+                "Cannot convert simple column. Expect data type {:?}, found {:?}",
+                T::data_type(),
+                col.data_type
+            )));
+        }
 
-        Ok(Self {
-            desc: !sort_columns_descriptions[0].asc,
-            _t: PhantomData,
+        let col = match &col.value {
+            Value::Scalar(v) => {
+                let builder = ColumnBuilder::repeat(&v.as_ref(), num_rows, &col.data_type);
+                builder.build()
+            }
+            Value::Column(c) => c.clone(),
+        };
+
+        Ok(SimpleRowsAsc {
+            inner: T::try_downcast_column(&col).unwrap(),
         })
     }
+}
 
-    fn convert(&mut self, columns: &[BlockEntry], num_rows: usize) -> Result<SimpleRows<T>> {
+impl<T> RowConverter<SimpleRowsDesc<T>> for SimpleRowConverter<T>
+where
+    T: ArgType,
+    for<'a> T::ScalarRef<'a>: Ord,
+{
+    fn create(
+        sort_columns_descriptions: &[SortColumnDescription],
+        _: DataSchemaRef,
+    ) -> Result<Self> {
+        assert!(sort_columns_descriptions.len() == 1);
+        assert!(!sort_columns_descriptions[0].asc);
+        Ok(Self { _t: PhantomData })
+    }
+
+    fn convert(&mut self, columns: &[BlockEntry], num_rows: usize) -> Result<SimpleRowsDesc<T>> {
         assert!(columns.len() == 1);
         let col = &columns[0];
         if col.data_type != T::data_type() {
@@ -175,10 +186,8 @@ where
             Value::Column(c) => c.clone(),
         };
 
-        let rows = SimpleRows {
+        Ok(SimpleRowsDesc {
             inner: T::try_downcast_column(&col).unwrap(),
-            desc: self.desc,
-        };
-        Ok(rows)
+        })
     }
 }