feat: date/timestamp bound check

Cargo.toml

@@ -67,3 +67,4 @@ rustc-demangle = { opt-level = 3 }
 
 [patch.crates-io]
 parquet2 = { version = "0.10", optional = true, git = "https://github.com/datafuse-extras/parquet2", rev = "daae989" }
+chrono = { git = "https://github.com/datafuse-extras/chrono", rev = "279f590" }

common/datavalues/src/types/date_converter.rs

@@ -39,11 +39,11 @@ where T: AsPrimitive<i64>
     }
 
     fn to_date_time64(&self, precision: usize, tz: &Tz) -> DateTime<Tz> {
-        let nano = self.as_() * 10i64.pow(9 - precision as u32);
+        let micros = self.as_() * 10i64.pow(6 - precision as u32);
 
-        let sec = nano / 1_000_000_000;
-        let nsec = nano % 1_000_000_000;
+        let sec = micros / 1_000_000;
+        let nanos = micros % 1_000_000 * 1000;
 
-        tz.timestamp(sec, nsec as u32)
+        tz.timestamp(sec, nanos as u32)
     }
 }

common/datavalues/src/types/deserializations/date.rs

@@ -35,6 +35,7 @@ where
 {
     fn de_binary(&mut self, reader: &mut &[u8]) -> Result<()> {
         let value: T = reader.read_scalar()?;
+        let _ = check_date(value.as_i32())?;
         self.builder.append_value(value);
         Ok(())
     }
@@ -47,6 +48,7 @@ where
         for row in 0..rows {
             let mut reader = &reader[step * row..];
             let value: T = reader.read_scalar()?;
+            let _ = check_date(value.as_i32())?;
             self.builder.append_value(value);
         }
         Ok(())
@@ -57,7 +59,9 @@ where
             serde_json::Value::String(v) => {
                 let mut reader = BufferReader::new(v.as_bytes());
                 let date = reader.read_date_text()?;
-                self.builder.append_value(uniform(date));
+                let days = uniform(date);
+                let _ = check_date(days.as_i32())?;
+                self.builder.append_value(days);
                 Ok(())
             }
             _ => Err(ErrorCode::BadBytes("Incorrect boolean value")),
@@ -67,47 +71,58 @@ where
     fn de_whole_text(&mut self, reader: &[u8]) -> Result<()> {
         let mut reader = BufferReader::new(reader);
         let date = reader.read_date_text()?;
+        let days = uniform(date);
+        let _ = check_date(days.as_i32())?;
         reader.must_eof()?;
-        self.builder.append_value(uniform(date));
+        self.builder.append_value(days);
         Ok(())
     }
 
     fn de_text_quoted(&mut self, reader: &mut CpBufferReader) -> Result<()> {
         reader.must_ignore_byte(b'\'')?;
         let date = reader.read_date_text()?;
+        let days = uniform(date);
+        let _ = check_date(days.as_i32())?;
         reader.must_ignore_byte(b'\'')?;
 
-        self.builder.append_value(uniform(date));
+        self.builder.append_value(days);
         Ok(())
     }
 
     fn de_text(&mut self, reader: &mut CpBufferReader) -> Result<()> {
         let date = reader.read_date_text()?;
-        self.builder.append_value(uniform(date));
+        let days = uniform(date);
+        let _ = check_date(days.as_i32())?;
+        self.builder.append_value(days);
         Ok(())
     }
 
     fn de_text_csv(&mut self, reader: &mut CpBufferReader) -> Result<()> {
         let maybe_quote = reader.ignore(|f| f == b'\'' || f == b'"')?;
         let date = reader.read_date_text()?;
+        let days = uniform(date);
+        let _ = check_date(days.as_i32())?;
         if maybe_quote {
             reader.must_ignore(|f| f == b'\'' || f == b'"')?;
         }
-        self.builder.append_value(uniform(date));
+        self.builder.append_value(days);
         Ok(())
     }
 
     fn de_text_json(&mut self, reader: &mut CpBufferReader) -> Result<()> {
         reader.must_ignore_byte(b'"')?;
         let date = reader.read_date_text()?;
+        let days = uniform(date);
+        let _ = check_date(days.as_i32())?;
         reader.must_ignore_byte(b'"')?;
 
-        self.builder.append_value(uniform(date));
+        self.builder.append_value(days);
         Ok(())
     }
 
     fn append_data_value(&mut self, value: DataValue) -> Result<()> {
         let v = value.as_i64()? as i32;
+        let _ = check_date(v)?;
         self.builder.append_value(v.as_());
         Ok(())
     }

common/datavalues/src/types/deserializations/timestamp.rs

@@ -37,6 +37,7 @@ where
 {
     fn de_binary(&mut self, reader: &mut &[u8]) -> Result<()> {
         let value: T = reader.read_scalar()?;
+        let _ = check_timestamp(value.as_i64())?;
         self.builder.append_value(value);
         Ok(())
     }
@@ -49,6 +50,7 @@ where
         for row in 0..rows {
             let mut reader = &reader[step * row..];
             let value: T = reader.read_scalar()?;
+            let _ = check_timestamp(value.as_i64())?;
             self.builder.append_value(value);
         }
         Ok(())
@@ -60,8 +62,9 @@ where
                 let v = v.clone();
                 let mut reader = BufferReader::new(v.as_bytes());
                 let ts = reader.read_timestamp_text(&self.tz)?;
-                self.builder
-                    .append_value(uniform(ts.timestamp_nanos(), self.precision).as_());
+                let micros = uniform(ts.timestamp_micros(), self.precision);
+                let _ = check_timestamp(micros)?;
+                self.builder.append_value(micros.as_());
                 Ok(())
             }
             _ => Err(ErrorCode::BadBytes("Incorrect boolean value")),
@@ -71,52 +74,57 @@ where
     fn de_text_quoted(&mut self, reader: &mut CpBufferReader) -> Result<()> {
         reader.must_ignore_byte(b'\'')?;
         let ts = reader.read_timestamp_text(&self.tz)?;
+        let micros = uniform(ts.timestamp_micros(), self.precision);
+        let _ = check_timestamp(micros)?;
         reader.must_ignore_byte(b'\'')?;
-
-        self.builder
-            .append_value(uniform(ts.timestamp_nanos(), self.precision).as_());
+        self.builder.append_value(micros.as_());
         Ok(())
     }
 
     fn de_whole_text(&mut self, reader: &[u8]) -> Result<()> {
         let mut reader = BufferReader::new(reader);
         let ts = reader.read_timestamp_text(&self.tz)?;
+        let micros = uniform(ts.timestamp_micros(), self.precision);
+        let _ = check_timestamp(micros)?;
         reader.must_eof()?;
-        self.builder
-            .append_value(uniform(ts.timestamp_nanos(), self.precision).as_());
+        self.builder.append_value(micros.as_());
         Ok(())
     }
 
     fn de_text(&mut self, reader: &mut CpBufferReader) -> Result<()> {
         let ts = reader.read_timestamp_text(&self.tz)?;
-        self.builder
-            .append_value(uniform(ts.timestamp_nanos(), self.precision).as_());
+        let micros = uniform(ts.timestamp_micros(), self.precision);
+        let _ = check_timestamp(micros)?;
+        self.builder.append_value(micros.as_());
         Ok(())
     }
 
     fn de_text_csv(&mut self, reader: &mut CpBufferReader) -> Result<()> {
         let maybe_quote = reader.ignore(|f| f == b'\'' || f == b'"')?;
         let ts = reader.read_timestamp_text(&self.tz)?;
+        let micros = uniform(ts.timestamp_micros(), self.precision);
+        let _ = check_timestamp(micros)?;
         if maybe_quote {
             reader.must_ignore(|f| f == b'\'' || f == b'"')?;
         }
-        self.builder
-            .append_value(uniform(ts.timestamp_nanos(), self.precision).as_());
+        self.builder.append_value(micros.as_());
         Ok(())
     }
 
     fn de_text_json(&mut self, reader: &mut CpBufferReader) -> Result<()> {
         reader.must_ignore_byte(b'"')?;
         let ts = reader.read_timestamp_text(&self.tz)?;
+        let micros = uniform(ts.timestamp_micros(), self.precision);
+        let _ = check_timestamp(micros)?;
         reader.must_ignore_byte(b'"')?;
 
-        self.builder
-            .append_value(uniform(ts.timestamp_nanos(), self.precision).as_());
+        self.builder.append_value(micros.as_());
         Ok(())
     }
 
     fn append_data_value(&mut self, value: DataValue) -> Result<()> {
         let v = value.as_i64()?;
+        let _ = check_timestamp(v)?;
         self.builder.append_value(v.as_());
         Ok(())
     }
@@ -131,11 +139,6 @@ where
 }
 
 #[inline]
-fn uniform(nanos: i64, precision: usize) -> i64 {
-    match precision {
-        0 => nanos as i64 / 1_000_000_000,
-        3 => nanos as i64 / 1_000_000,
-        6 => nanos as i64 / 1_000,
-        _ => nanos as i64,
-    }
+fn uniform(micros: i64, precision: usize) -> i64 {
+    micros / 10_i64.pow(6 - precision as u32)
 }

common/datavalues/src/types/type_date.rs

@@ -16,12 +16,29 @@ use std::collections::BTreeMap;
 use std::sync::Arc;
 
 use common_arrow::arrow::datatypes::DataType as ArrowType;
+use common_exception::ErrorCode;
 use common_exception::Result;
 
 use super::data_type::DataType;
 use super::type_id::TypeID;
 use crate::prelude::*;
 
+/// date ranges from 1000-01-01 to 9999-12-31
+/// date_max and date_min means days offset from 1970-01-01
+/// any date not in the range will be invalid
+pub const DATE_MAX: i32 = 2932896;
+pub const DATE_MIN: i32 = -354285;
+
+#[inline]
+pub fn check_date(days: i32) -> Result<()> {
+    if days >= DATE_MIN && days <= DATE_MAX {
+        return Ok(());
+    }
+    Err(ErrorCode::InvalidDate(
+        "Date only ranges from 1000-01-01 to 9999-12-31",
+    ))
+}
+
 #[derive(Default, Clone, serde::Deserialize, serde::Serialize)]
 pub struct DateType {}
 

common/datavalues/src/types/type_factory.rs

@@ -52,7 +52,7 @@ static TYPE_FACTORY: Lazy<Arc<TypeFactory>> = Lazy::new(|| {
 
     // Timestamp is a special case
     {
-        for precision in 0..10 {
+        for precision in 0..7 {
             type_factory.register(TimestampType::arc(precision, None));
         }
     }

common/datavalues/src/types/type_id.rs

@@ -51,7 +51,7 @@ pub enum TypeID {
     Date,
 
     /// A 64-bit timestamp representing the elapsed time since UNIX epoch (1970-01-01)
-    /// in nanoseconds, it's physical type is Int64
+    /// in microseconds, it's physical type is Int64
     /// store UTC timestamp
     Timestamp,
 

common/datavalues/src/types/type_timestamp.rs

@@ -20,6 +20,7 @@ use chrono::TimeZone;
 use chrono::Utc;
 use chrono_tz::Tz;
 use common_arrow::arrow::datatypes::DataType as ArrowType;
+use common_exception::ErrorCode;
 use common_exception::Result;
 
 use super::data_type::DataType;
@@ -28,10 +29,26 @@ use super::data_type::ARROW_EXTENSION_NAME;
 use super::type_id::TypeID;
 use crate::prelude::*;
 
+/// timestamp ranges from 1000-01-01 00:00:00.000000 to 9999-12-31 23:59:59.999999
+/// timestamp_max and timestamp_min means days offset from 1970-01-01 00:00:00.000000
+/// any timestamp not in the range will be invalid
+pub const TIMESTAMP_MAX: i64 = 253402300799999999;
+pub const TIMESTAMP_MIN: i64 = -30610224000000000;
+
+#[inline]
+pub fn check_timestamp(micros: i64) -> Result<()> {
+    if micros >= TIMESTAMP_MIN && micros <= TIMESTAMP_MAX {
+        return Ok(());
+    }
+    Err(ErrorCode::InvalidTimestamp(
+        "Timestamp only ranges from 1000-01-01 00:00:00.000000 to 9999-12-31 23:59:59.999999",
+    ))
+}
+
 #[derive(Default, Clone, serde::Deserialize, serde::Serialize)]
 pub struct TimestampType {
     /// The time resolution is determined by the precision parameter, range from 0 to 9
-    /// Typically are used - 0 (seconds) 3 (milliseconds), 6 (microseconds), 9 (nanoseconds).
+    /// Typically are used - 0 (seconds) 3 (milliseconds), 6 (microseconds)
     precision: usize,
     /// tz indicates the timezone, if it's None, it's UTC.
     tz: Option<String>,
@@ -56,21 +73,21 @@ impl TimestampType {
 
     #[inline]
     pub fn utc_timestamp(&self, v: i64) -> DateTime<Utc> {
-        let v = v * 10_i64.pow(9 - self.precision as u32);
+        let v = v * 10_i64.pow(6 - self.precision as u32);
 
         // ns
-        Utc.timestamp(v / 1_000_000_000, (v % 1_000_000_000) as u32)
+        Utc.timestamp(v / 1_000_000, (v % 1_000_000 * 1000) as u32)
     }
 
     #[inline]
     pub fn to_seconds(&self, v: i64) -> i64 {
-        let v = v * 10_i64.pow(9 - self.precision as u32);
-        v / 1_000_000_000
+        let v = v * 10_i64.pow(6 - self.precision as u32);
+        v / 1_000_000
     }
 
     #[inline]
-    pub fn from_nano_seconds(&self, v: i64) -> i64 {
-        v / 10_i64.pow(9 - self.precision as u32)
+    pub fn from_micro_seconds(&self, v: i64) -> i64 {
+        v / 10_i64.pow(6 - self.precision as u32)
     }
 
     pub fn format_string(&self) -> String {
@@ -106,9 +123,6 @@ impl DataType for TimestampType {
             4 => &["DateTime(4)"],
             5 => &["DateTime(5)"],
             6 => &["Timestamp", "DateTime"],
-            7 => &["DateTime(7)"],
-            8 => &["DateTime(8)"],
-            9 => &["DateTime(9)"],
             _ => &[],
         }
     }

common/datavalues/src/types/type_traits.rs

@@ -23,6 +23,7 @@ use crate::DataValue;
 use crate::DateType;
 use crate::Scalar;
 use crate::TimestampType;
+use crate::TypeID;
 use crate::VariantType;
 use crate::VariantValue;
 
@@ -86,9 +87,19 @@ pub trait FloatType: PrimitiveType {}
 impl FloatType for f32 {}
 impl FloatType for f64 {}
 
-pub trait LogicalDateType: PrimitiveType {}
-impl LogicalDateType for i32 {}
-impl LogicalDateType for i64 {}
+pub trait LogicalDateType: PrimitiveType {
+    fn get_type_id() -> TypeID;
+}
+impl LogicalDateType for i32 {
+    fn get_type_id() -> TypeID {
+        TypeID::Date
+    }
+}
+impl LogicalDateType for i64 {
+    fn get_type_id() -> TypeID {
+        TypeID::Timestamp
+    }
+}
 
 pub trait ToDateType {
     fn to_date_type() -> DataTypePtr;