feat: support interval comparison (risingwavelabs#3222)

1. fix timestamp substract timestamp.
2. support interval comparison. From pgsql, 1 month equal to 30 days and 1 day equal to 86400000 ms.

e2e_test/batch/types/interval.slt.part

@@ -31,4 +31,24 @@ SELECT interval '5 minute', interval '2 m';
 query TTTTT
 SELECT interval '6 second';
 ----
-00:00:06
+00:00:06
+
+query T
+SELECT interval '1' month = interval '30' day;
+----
+t
+
+query T
+SELECT interval '1' day = interval '24' hour;
+----
+t
+
+query T
+SELECT interval '1' day = interval '86400' second;
+----
+t
+
+query T
+SELECT interval '1' day - interval '12' hour = interval '12' hour;
+----
+t

e2e_test/batch/types/time.slt.part

@@ -2,3 +2,8 @@ query T
 values(extract(hour from timestamp '2001-02-16 20:38:40'));
 ----
 20
+
+query TTTTT
+select timestamp '2001-03-16 23:38:45' - timestamp '2001-02-16 20:38:40';
+----
+28 days 03:00:05

src/common/src/types/interval.rs

@@ -12,7 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::cmp::Ordering;
 use std::fmt::{Display, Formatter};
+use std::hash::{Hash, Hasher};
 use std::io::Write;
 use std::ops::{Add, Sub};
 
@@ -34,17 +36,16 @@ use super::*;
 /// One month may contain 28/31 days. One day may contain 23/25 hours.
 /// This internals is learned from PG:
 /// <https://www.postgresql.org/docs/9.1/datatype-datetime.html#:~:text=field%20is%20negative.-,Internally,-interval%20values%20are>
-///
-/// FIXME: if this derives `PartialEq` and `PartialOrd`, caller must guarantee the fields are valid.
-#[derive(
-    Debug, Clone, Copy, Eq, PartialEq, PartialOrd, Ord, Hash, Default, Serialize, Deserialize,
-)]
+/// FIXME: the comparison of memcomparable encoding will be just compare these three numbers.
+#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize)]
 pub struct IntervalUnit {
     months: i32,
     days: i32,
     ms: i64,
 }
 
+const DAY_MS: i64 = 86400000;
+
 impl IntervalUnit {
     pub fn new(months: i32, days: i32, ms: i64) -> Self {
         IntervalUnit { months, days, ms }
@@ -95,6 +96,17 @@ impl IntervalUnit {
         }
     }
 
+    /// Justify interval, convert 1 month to 30 days and 86400 ms to 1 day.
+    /// If day is positive, complement the ms negative value.
+    /// These rules only use in interval comparison.
+    pub fn justify_interval(&mut self) {
+        let month = (self.months * 30) as i64 * DAY_MS;
+        self.ms = self.ms + month + (self.days) as i64 * DAY_MS;
+        self.days = (self.ms / DAY_MS) as i32;
+        self.ms %= DAY_MS;
+        self.months = 0;
+    }
+
     #[must_use]
     pub fn negative(&self) -> Self {
         IntervalUnit {
@@ -236,6 +248,46 @@ impl Add for IntervalUnit {
     }
 }
 
+impl PartialOrd for IntervalUnit {
+    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+        if self.eq(other) {
+            Some(Ordering::Equal)
+        } else {
+            let diff = *self - *other;
+            let days = (diff.months * 30 + diff.days) as i64;
+            Some((days * DAY_MS + diff.ms).cmp(&0))
+        }
+    }
+}
+
+impl Hash for IntervalUnit {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        let mut interval = *self;
+        interval.justify_interval();
+        interval.months.hash(state);
+        interval.ms.hash(state);
+        interval.days.hash(state);
+    }
+}
+
+impl PartialEq for IntervalUnit {
+    fn eq(&self, other: &Self) -> bool {
+        let mut interval = *self;
+        interval.justify_interval();
+        let mut other = *other;
+        other.justify_interval();
+        interval.days == other.days && interval.ms == other.ms
+    }
+}
+
+impl Eq for IntervalUnit {}
+
+impl Ord for IntervalUnit {
+    fn cmp(&self, other: &Self) -> Ordering {
+        self.partial_cmp(other).unwrap()
+    }
+}
+
 impl CheckedAdd for IntervalUnit {
     fn checked_add(&self, other: &Self) -> Option<Self> {
         let months = self.months.checked_add(other.months)?;

src/expr/src/expr/template.rs

@@ -382,6 +382,7 @@ macro_rules! for_all_cmp_variants {
             { float32, decimal, float64, $general_f },
             { float64, decimal, float64, $general_f },
             { timestamp, timestamp, timestamp, $general_f },
+            { interval, interval, interval, $general_f },
             { date, date, date, $general_f },
             { boolean, boolean, boolean, $general_f },
             { timestamp, date, timestamp, $general_f },

src/expr/src/vector_op/arithmetic_op.rs

@@ -15,7 +15,9 @@
 use std::any::type_name;
 use std::convert::TryInto;
 use std::fmt::Debug;
+use std::ops::Sub;
 
+use chrono::Duration;
 use num_traits::{CheckedDiv, CheckedMul, CheckedNeg, CheckedRem, CheckedSub, Signed};
 use risingwave_common::types::{
     CheckedAdd, Decimal, IntervalUnit, NaiveDateTimeWrapper, NaiveDateWrapper,
@@ -125,7 +127,9 @@ pub fn timestamp_timestamp_sub<T1, T2, T3>(
     r: NaiveDateTimeWrapper,
 ) -> Result<IntervalUnit> {
     let tmp = l.0 - r.0;
-    Ok(IntervalUnit::new(0, tmp.num_days() as i32, 0))
+    let days = tmp.num_days();
+    let ms = tmp.sub(Duration::days(tmp.num_days())).num_milliseconds();
+    Ok(IntervalUnit::new(0, days as i32, ms))
 }
 
 #[inline(always)]