fix(query): interval order by is incorrect

Cargo.toml

@@ -419,6 +419,7 @@ prost-build = { version = "0.13" }
 prqlc = "0.11.3"
 raft-log = { version = "0.2.6" }
 rand = { version = "0.8.5", features = ["small_rng"] }
+rand_distr = "0.4.3"
 rayon = "1.9.0"
 recursive = "0.1.1"
 redis = { version = "0.27.5", features = ["tokio-comp", "connection-manager"] }

src/common/column/src/types/native.rs

@@ -268,9 +268,6 @@ impl NativeType for days_ms {
 #[repr(C)]
 pub struct months_days_micros(pub i128);
 
-const MICROS_PER_DAY: i64 = 24 * 3600 * 1_000_000;
-const MICROS_PER_MONTH: i64 = 30 * MICROS_PER_DAY;
-
 impl Hash for months_days_micros {
     fn hash<H: Hasher>(&self, state: &mut H) {
         self.total_micros().hash(state)
@@ -296,6 +293,9 @@ impl Ord for months_days_micros {
 }
 
 impl months_days_micros {
+    pub const MICROS_PER_DAY: i64 = 24 * 3600 * 1_000_000;
+    pub const MICROS_PER_MONTH: i64 = 30 * Self::MICROS_PER_DAY;
+
     pub fn new(months: i32, days: i32, microseconds: i64) -> Self {
         let months_bits = (months as i128) << 96;
         // converting to u32 before i128 ensures we’re working with the raw, unsigned bit pattern of the i32 value,
@@ -320,21 +320,29 @@ impl months_days_micros {
     }
 
     pub fn total_micros(&self) -> i64 {
-        let months_micros = (self.months() as i64).checked_mul(MICROS_PER_MONTH);
-        let days_micros = (self.days() as i64).checked_mul(MICROS_PER_DAY);
-
-        months_micros
-            .and_then(|months_micros| days_micros.map(|days_micros| months_micros + days_micros))
-            .and_then(|total_micros| total_micros.checked_add(self.microseconds()))
-            .unwrap_or_else(|| {
-                error!(
-                    "interval is out of range: months={}, days={}, micros={}",
-                    self.months(),
-                    self.days(),
-                    self.microseconds()
-                );
-                0
-            })
+        self.try_total_micros().unwrap_or_else(|| {
+            error!(
+                "interval is out of range: months={}, days={}, micros={}",
+                self.months(),
+                self.days(),
+                self.microseconds()
+            );
+            0
+        })
+    }
+
+    pub fn try_total_micros(&self) -> Option<i64> {
+        [
+            (self.months() as i64).checked_mul(Self::MICROS_PER_MONTH),
+            (self.days() as i64).checked_mul(Self::MICROS_PER_DAY),
+            Some(self.microseconds()),
+        ]
+        .into_iter()
+        .reduce(|acc, x| match (acc, x) {
+            (Some(acc), Some(x)) => acc.checked_add(x),
+            (acc, None) => acc,
+            (None, x) => x,
+        })?
     }
 }
 

src/query/expression/Cargo.toml

@@ -53,6 +53,7 @@ micromarshal = { workspace = true }
 num-bigint = { workspace = true }
 num-traits = { workspace = true }
 rand = { workspace = true }
+rand_distr = { workspace = true }
 recursive = { workspace = true }
 roaring = { workspace = true, features = ["serde"] }
 rust_decimal = { workspace = true }

src/query/expression/src/row/fixed.rs

@@ -14,7 +14,6 @@
 
 use databend_common_column::bitmap::Bitmap;
 use databend_common_column::types::months_days_micros;
-use databend_common_column::types::NativeType;
 use ethnum::i256;
 
 use super::row_converter::null_sentinel;
@@ -101,10 +100,10 @@ impl FixedLengthEncoding for F64 {
 }
 
 impl FixedLengthEncoding for months_days_micros {
-    type Encoded = [u8; 16];
+    type Encoded = [u8; 8];
 
-    fn encode(self) -> [u8; 16] {
-        self.to_be_bytes()
+    fn encode(self) -> [u8; 8] {
+        self.total_micros().encode()
     }
 }
 

src/query/expression/src/types/interval.rs

@@ -166,7 +166,7 @@ impl ValueType for IntervalType {
 
     #[inline(always)]
     fn compare(lhs: Self::ScalarRef<'_>, rhs: Self::ScalarRef<'_>) -> Ordering {
-        lhs.0.cmp(&rhs.0)
+        lhs.cmp(&rhs)
     }
 
     #[inline(always)]

src/query/expression/src/values.rs

@@ -1325,17 +1325,20 @@ impl Column {
                     .map(|_| rng.gen_range(DATE_MIN..=DATE_MAX))
                     .collect::<Vec<i32>>(),
             ),
-            DataType::Interval => IntervalType::from_data(
-                (0..len)
-                    .map(|_| {
-                        months_days_micros::new(
-                            rng.gen::<i32>(),
-                            rng.gen::<i32>(),
-                            rng.gen::<i64>(),
-                        )
-                    })
-                    .collect::<Vec<months_days_micros>>(),
-            ),
+            DataType::Interval => IntervalType::from_data(Vec::from_iter(
+                std::iter::repeat_with(|| {
+                    let normal = rand_distr::Normal::new(0.001, 1.0).unwrap();
+                    const MAX_MONTHS: i64 = i64::MAX / months_days_micros::MICROS_PER_MONTH;
+                    const MAX_DAYS: i64 = i64::MAX / months_days_micros::MICROS_PER_DAY;
+                    months_days_micros::new(
+                        (rng.sample(normal) * 0.3 * MAX_MONTHS as f64) as i32,
+                        (rng.sample(normal) * 0.3 * MAX_DAYS as f64) as i32,
+                        (rng.sample(normal) * 2.0 * months_days_micros::MICROS_PER_DAY as f64)
+                            as i64,
+                    )
+                })
+                .take(len),
+            )),
             DataType::Nullable(ty) => NullableColumn::new_column(
                 Column::random(ty, len, options),
                 Bitmap::from((0..len).map(|_| rng.gen_bool(0.5)).collect::<Vec<bool>>()),

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

@@ -141,6 +141,10 @@ where
             }
         }
 
+        if self.is_finished() {
+            return Ok(vec![]);
+        }
+
         Ok(self.build_task())
     }
 
@@ -160,6 +164,7 @@ where
 
     fn build_task(&mut self) -> Vec<DataBlock> {
         let partition = self.calc_partition_point();
+        assert!(partition.total > 0);
 
         let id = self.next_task_id();
         self.total_rows += partition.total;

tests/sqllogictests/suites/base/11_data_type/11_0007_data_type_interval.test

@@ -0,0 +1,10 @@
+statement ok
+create table test_r (weeks int16 not null,hours int16 not null) engine=random;
+
+query I
+with t as ( select to_weeks(weeks) + to_hours(hours) as i, weeks*168+hours as h from test_r limit 10000), t1 as ( select ROW_NUMBER() OVER (ORDER BY h) AS num_h,ROW_NUMBER() OVER (ORDER BY i) AS num_i from t) select count_if(num_h=num_i) from t1;
+----
+10000
+
+statement ok
+drop table test_r;