refactor(types): spilt timestamp to timestamp and timestampnano

e2e_test/batch/types/timestamp_ns.slt.part

@@ -2,7 +2,7 @@ statement ok
 SET RW_IMPLICIT_FLUSH TO true;
 
 statement ok
-create table t1(v1 int, v2 timestamp);
+create table t1(v1 int, v2 timestamp_ns);
 
 statement ok
 insert into t1 values(1,'2013-01-01 01:01:01.123456789'),(2,'2012-01-01 01:01:01.123456'),(3,'0000-01-01 01:01:01.123456789'),(4,'2213-01-01 01:01:01.123456789'),(5,null),(6,'2013-01-01 01:01:01.123456789');
@@ -62,7 +62,7 @@ select * from t1 where v2 >= '2012-01-01 01:01:01.123456';
 6  2013-01-01 01:01:01.123456789
 
 query T rowsort
-select v1, cast(v2 as date) as date_v2, cast(v2 as timestamp with time zone) as timestamptz_v2 from t1;
+select v1, cast(v2 as date) as date_v2, cast(v2 as timestamptz) as timestamptz_v2 from t1;
 ----
 1  2013-01-01     2013-01-01 01:01:01.123456+00:00
 2  2012-01-01     2012-01-01 01:01:01.123456+00:00
@@ -102,7 +102,7 @@ select v1, to_char(v2, 'YYYY-MM-DD HH24:MI:SS.NS') as formatted_v2 from t1;
 6  2013-01-01 01:01:01.123456789
 
 query T rowsort
-select generate_series('2013-01-01 01:01:01.123456789'::timestamp,'2013-01-01 01:01:05.123456790'::timestamp, '1 s');
+select generate_series('2013-01-01 01:01:01.123456789'::timestamp_ns,'2013-01-01 01:01:05.123456790'::timestamp_ns, '1 s');
 ----
 2013-01-01 01:01:01.123456789
 2013-01-01 01:01:02.123456789
@@ -194,7 +194,7 @@ select v1, extract(nanosecond from v2) from t1;
 6  1123456789
 
 query T rowsort
-select make_timestamp(2013, 01, 01, 01, 01, 1.123456789);
+select make_timestamp_ns(2013, 01, 01, 01, 01, 1.123456789);
 ----
 2013-01-01 01:01:01.123456789
 

e2e_test/iceberg/test_case/iceberg_engine.slt

@@ -36,6 +36,7 @@ v_bool boolean,
 v_date date,
 v_timestamp timestamptz,
 v_ts_ntz timestamp,
+v_timestamp_ns timestamp_ns,
 v_decimal decimal,
 v_map map(int, int),
 v_array int[],

proto/data.proto

@@ -53,6 +53,7 @@ message DataType {
     SERIAL = 19;
     INT256 = 20;
     MAP = 21;
+    TIMESTAMP_NANOSECOND = 22;
   }
   TypeName type_name = 1;
   // Data length for char.
@@ -104,6 +105,7 @@ enum ArrayType {
   SERIAL = 17;
   INT256 = 18;
   MAP = 20;
+  TIMESTAMP_NANOSECOND = 21;
 }
 
 message Array {

proto/expr.proto

@@ -56,6 +56,7 @@ message ExprNode {
     MAKE_DATE = 113;
     MAKE_TIME = 114;
     MAKE_TIMESTAMP = 115;
+    MAKE_TIMESTAMP_NS = 116;
     // From f64 to timestamp.
     // e.g. `select to_timestamp(1672044740.0)`
     SEC_TO_TIMESTAMPTZ = 104;

src/batch/executors/src/executor/postgres_query.rs

@@ -85,6 +85,7 @@ fn postgres_cell_to_scalar_impl(
         | DataType::Date
         | DataType::Time
         | DataType::Timestamp
+        | DataType::TimestampNanosecond
         | DataType::Timestamptz
         | DataType::Jsonb
         | DataType::Interval

src/common/src/array/arrow/arrow_iceberg.rs

@@ -97,6 +97,7 @@ impl ToArrow for IcebergArrowConvert {
             DataType::Date => self.date_type_to_arrow(),
             DataType::Time => self.time_type_to_arrow(),
             DataType::Timestamp => self.timestamp_type_to_arrow(),
+            DataType::TimestampNanosecond => self.timestampns_type_to_arrow(),
             DataType::Timestamptz => self.timestamptz_type_to_arrow(),
             DataType::Interval => self.interval_type_to_arrow(),
             DataType::Varchar => self.varchar_type_to_arrow(),
@@ -246,6 +247,7 @@ impl ToArrow for IcebergCreateTableArrowConvert {
             DataType::Date => self.date_type_to_arrow(),
             DataType::Time => self.time_type_to_arrow(),
             DataType::Timestamp => self.timestamp_type_to_arrow(),
+            DataType::TimestampNanosecond => self.timestampns_type_to_arrow(),
             DataType::Timestamptz => self.timestamptz_type_to_arrow(),
             DataType::Interval => self.interval_type_to_arrow(),
             DataType::Varchar => self.varchar_type_to_arrow(),

src/common/src/array/arrow/arrow_impl.rs

@@ -107,6 +107,7 @@ pub trait ToArrow {
             ArrayImpl::Date(array) => self.date_to_arrow(array),
             ArrayImpl::Time(array) => self.time_to_arrow(array),
             ArrayImpl::Timestamp(array) => self.timestamp_to_arrow(array),
+            ArrayImpl::TimestampNanosecond(array) => self.timestampns_to_arrow(array),
             ArrayImpl::Timestamptz(array) => self.timestamptz_to_arrow(array),
             ArrayImpl::Interval(array) => self.interval_to_arrow(array),
             ArrayImpl::Utf8(array) => self.utf8_to_arrow(array),
@@ -180,6 +181,14 @@ pub trait ToArrow {
         )))
     }
 
+    #[inline]
+    fn timestampns_to_arrow(
+        &self,
+        array: &TimestampNanosecondArray,
+    ) -> Result<arrow_array::ArrayRef, ArrayError> {
+        Ok(Arc::new(arrow_array::TimestampNanosecondArray::from(array)))
+    }
+
     #[inline]
     fn timestamptz_to_arrow(
         &self,
@@ -319,6 +328,7 @@ pub trait ToArrow {
             DataType::Date => self.date_type_to_arrow(),
             DataType::Time => self.time_type_to_arrow(),
             DataType::Timestamp => self.timestamp_type_to_arrow(),
+            DataType::TimestampNanosecond => self.timestampns_type_to_arrow(),
             DataType::Timestamptz => self.timestamptz_type_to_arrow(),
             DataType::Interval => self.interval_type_to_arrow(),
             DataType::Varchar => self.varchar_type_to_arrow(),
@@ -382,6 +392,10 @@ pub trait ToArrow {
     fn timestamp_type_to_arrow(&self) -> arrow_schema::DataType {
         arrow_schema::DataType::Timestamp(arrow_schema::TimeUnit::Microsecond, None)
     }
+    #[inline]
+    fn timestampns_type_to_arrow(&self) -> arrow_schema::DataType {
+        arrow_schema::DataType::Timestamp(arrow_schema::TimeUnit::Nanosecond, None)
+    }
 
     #[inline]
     fn timestamptz_type_to_arrow(&self) -> arrow_schema::DataType {
@@ -523,7 +537,7 @@ pub trait FromArrow {
             Timestamp(Second, Some(_)) => DataType::Timestamptz,
             Timestamp(Millisecond, None) => DataType::Timestamp,
             Timestamp(Millisecond, Some(_)) => DataType::Timestamptz,
-            Timestamp(Nanosecond, None) => DataType::Timestamp,
+            Timestamp(Nanosecond, None) => DataType::TimestampNanosecond,
             Timestamp(Nanosecond, Some(_)) => DataType::Timestamptz,
             Interval(MonthDayNano) => DataType::Interval,
             Utf8 => DataType::Varchar,
@@ -801,7 +815,7 @@ pub trait FromArrow {
         &self,
         array: &arrow_array::TimestampNanosecondArray,
     ) -> Result<ArrayImpl, ArrayError> {
-        Ok(ArrayImpl::Timestamp(array.into()))
+        Ok(ArrayImpl::TimestampNanosecond(array.into()))
     }
 
     fn from_timestampns_some_array(
@@ -1046,6 +1060,11 @@ converts_with_timeunit!(TimestamptzArray, arrow_array::TimestampMillisecondArray
 converts_with_timeunit!(TimestamptzArray, arrow_array::TimestampMicrosecondArray, TimeUnit::Microsecond, @map);
 converts_with_timeunit!(TimestamptzArray, arrow_array::TimestampNanosecondArray, TimeUnit::Nanosecond, @map);
 
+converts_with_timeunit!(TimestampNanosecondArray, arrow_array::TimestampSecondArray, TimeUnit::Second, @map);
+converts_with_timeunit!(TimestampNanosecondArray, arrow_array::TimestampMillisecondArray, TimeUnit::Millisecond, @map);
+converts_with_timeunit!(TimestampNanosecondArray, arrow_array::TimestampMicrosecondArray, TimeUnit::Microsecond, @map);
+converts_with_timeunit!(TimestampNanosecondArray, arrow_array::TimestampNanosecondArray, TimeUnit::Nanosecond, @map);
+
 /// Converts RisingWave value from and into Arrow value.
 trait FromIntoArrow {
     /// The corresponding element type in the Arrow array.
@@ -1162,6 +1181,37 @@ impl FromIntoArrowWithUnit for Timestamp {
     }
 }
 
+impl FromIntoArrowWithUnit for TimestampNanosecond {
+    type ArrowType = i64;
+    type TimestampType = TimeUnit;
+
+    fn from_arrow_with_unit(value: Self::ArrowType, time_unit: Self::TimestampType) -> Self {
+        match time_unit {
+            TimeUnit::Second => {
+                TimestampNanosecond(DateTime::from_timestamp(value as _, 0).unwrap().naive_utc())
+            }
+            TimeUnit::Millisecond => {
+                TimestampNanosecond(DateTime::from_timestamp_millis(value).unwrap().naive_utc())
+            }
+            TimeUnit::Microsecond => {
+                TimestampNanosecond(DateTime::from_timestamp_micros(value).unwrap().naive_utc())
+            }
+            TimeUnit::Nanosecond => {
+                TimestampNanosecond(DateTime::from_timestamp_nanos(value).naive_utc())
+            }
+        }
+    }
+
+    fn into_arrow_with_unit(self, time_unit: Self::TimestampType) -> Self::ArrowType {
+        match time_unit {
+            TimeUnit::Second => self.0.and_utc().timestamp(),
+            TimeUnit::Millisecond => self.0.and_utc().timestamp_millis(),
+            TimeUnit::Microsecond => self.0.and_utc().timestamp_micros(),
+            TimeUnit::Nanosecond => self.0.and_utc().timestamp_nanos_opt().unwrap(),
+        }
+    }
+}
+
 impl FromIntoArrowWithUnit for Timestamptz {
     type ArrowType = i64;
     type TimestampType = TimeUnit;

src/common/src/array/chrono_array.rs

@@ -12,17 +12,19 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use super::{PrimitiveArray, PrimitiveArrayBuilder};
+use super::{PrimitiveArray, PrimitiveArrayBuilder, TimestampNanosecond};
 use crate::types::{Date, Time, Timestamp, Timestamptz};
 
 pub type DateArray = PrimitiveArray<Date>;
 pub type TimeArray = PrimitiveArray<Time>;
 pub type TimestampArray = PrimitiveArray<Timestamp>;
+pub type TimestampNanosecondArray = PrimitiveArray<TimestampNanosecond>;
 pub type TimestamptzArray = PrimitiveArray<Timestamptz>;
 
 pub type DateArrayBuilder = PrimitiveArrayBuilder<Date>;
 pub type TimeArrayBuilder = PrimitiveArrayBuilder<Time>;
 pub type TimestampArrayBuilder = PrimitiveArrayBuilder<Timestamp>;
+pub type TimestampNanosecondArrayBuilder = PrimitiveArrayBuilder<TimestampNanosecond>;
 pub type TimestamptzArrayBuilder = PrimitiveArrayBuilder<Timestamptz>;
 
 #[cfg(test)]

src/common/src/array/mod.rs

@@ -45,7 +45,8 @@ pub use bool_array::{BoolArray, BoolArrayBuilder};
 pub use bytes_array::*;
 pub use chrono_array::{
     DateArray, DateArrayBuilder, TimeArray, TimeArrayBuilder, TimestampArray,
-    TimestampArrayBuilder, TimestamptzArray, TimestamptzArrayBuilder,
+    TimestampArrayBuilder, TimestampNanosecondArray, TimestampNanosecondArrayBuilder,
+    TimestamptzArray, TimestamptzArrayBuilder,
 };
 pub use data_chunk::{DataChunk, DataChunkTestExt};
 pub use data_chunk_iter::RowRef;

src/common/src/array/primitive_array.rs

@@ -129,6 +129,7 @@ impl_primitive_for_others! {
     { Date, Date, Date },
     { Time, Time, Time },
     { Timestamp, Timestamp, Timestamp },
+    { TimestampNanosecond, TimestampNanosecond, TimestampNanosecond },
     { Timestamptz, Timestamptz, Timestamptz }
 }
 

src/common/src/array/proto_reader.rs

@@ -36,6 +36,9 @@ impl ArrayImpl {
             PbArrayType::Date => read_primitive_array::<Date>(array, cardinality)?,
             PbArrayType::Time => read_primitive_array::<Time>(array, cardinality)?,
             PbArrayType::Timestamp => read_primitive_array::<Timestamp>(array, cardinality)?,
+            PbArrayType::TimestampNanosecond => {
+                read_primitive_array::<TimestampNanosecond>(array, cardinality)?
+            }
             PbArrayType::Timestamptz => read_primitive_array::<Timestamptz>(array, cardinality)?,
             PbArrayType::Interval => read_primitive_array::<Interval>(array, cardinality)?,
             PbArrayType::Jsonb => JsonbArray::from_protobuf(array)?,

src/common/src/hash/key.rs

@@ -36,7 +36,7 @@ use static_assertions::const_assert_eq;
 use crate::array::{ListValue, MapValue, StructValue};
 use crate::types::{
     DataType, Date, Decimal, Int256, Int256Ref, JsonbVal, Scalar, ScalarRef, ScalarRefImpl, Serial,
-    Time, Timestamp, Timestamptz, F32, F64,
+    Time, Timestamp, TimestampNanosecond, Timestamptz, F32, F64,
 };
 use crate::util::hash_util::{Crc32FastBuilder, XxHash64Builder};
 use crate::util::sort_util::OrderType;
@@ -584,6 +584,25 @@ impl HashKeyDe for Timestamp {
     }
 }
 
+impl HashKeySer<'_> for TimestampNanosecond {
+    fn serialize_into(self, mut buf: impl BufMut) {
+        buf.put_i64_ne(self.0.and_utc().timestamp());
+        buf.put_u32_ne(self.0.and_utc().timestamp_subsec_nanos());
+    }
+
+    fn exact_size() -> Option<usize> {
+        Some(12)
+    }
+}
+
+impl HashKeyDe for TimestampNanosecond {
+    fn deserialize(_data_type: &DataType, mut buf: impl Buf) -> Self {
+        let secs = buf.get_i64_ne();
+        let nsecs = buf.get_u32_ne();
+        TimestampNanosecond::with_secs_nsecs(secs, nsecs).unwrap()
+    }
+}
+
 impl HashKeySer<'_> for Time {
     fn serialize_into(self, mut buf: impl BufMut) {
         buf.put_u32_ne(self.0.num_seconds_from_midnight());

src/common/src/row/owned_row.rs

@@ -18,7 +18,8 @@ use risingwave_common_estimate_size::EstimateSize;
 
 use super::Row;
 use crate::types::{
-    DataType, Date, Datum, DatumRef, Decimal, Interval, ScalarImpl, Time, Timestamp, ToDatumRef,
+    DataType, Date, Datum, DatumRef, Decimal, Interval, ScalarImpl, Time, Timestamp,
+    TimestampNanosecond, ToDatumRef,
 };
 use crate::util::iter_util::ZipEqDebug;
 use crate::util::value_encoding;
@@ -81,6 +82,9 @@ impl OwnedRow {
                     DataType::Date => x.parse::<Date>().unwrap().into(),
                     DataType::Time => x.parse::<Time>().unwrap().into(),
                     DataType::Timestamp => x.parse::<Timestamp>().unwrap().into(),
+                    DataType::TimestampNanosecond => {
+                        x.parse::<TimestampNanosecond>().unwrap().into()
+                    }
                     DataType::Interval => x.parse::<Interval>().unwrap().into(),
                     DataType::Decimal => x.parse::<Decimal>().unwrap().into(),
                     _ => todo!(),

src/common/src/test_utils/rand_array.rs

@@ -27,7 +27,7 @@ use rand::{Rng, SeedableRng};
 use crate::array::{Array, ArrayBuilder, ArrayRef, ListValue, MapValue, StructValue};
 use crate::types::{
     DataType, Date, Decimal, Int256, Interval, JsonbVal, MapType, NativeType, Scalar, Serial, Time,
-    Timestamp, Timestamptz,
+    Timestamp, TimestampNanosecond, Timestamptz,
 };
 
 pub trait RandValue {
@@ -106,6 +106,12 @@ impl RandValue for Timestamp {
     }
 }
 
+impl RandValue for TimestampNanosecond {
+    fn rand_value<R: Rng>(rand: &mut R) -> Self {
+        TimestampNanosecond::new(Date::rand_value(rand).0.and_time(Time::rand_value(rand).0))
+    }
+}
+
 impl RandValue for Timestamptz {
     fn rand_value<R: Rng>(rand: &mut R) -> Self {
         Timestamptz::from_micros(rand.gen())

src/common/src/test_utils/rand_chunk.rs

@@ -14,8 +14,8 @@
 
 use crate::array::{
     BoolArray, DataChunk, DateArray, DecimalArray, F32Array, F64Array, I16Array, I32Array,
-    I64Array, Int256Array, IntervalArray, SerialArray, TimeArray, TimestampArray, TimestamptzArray,
-    Utf8Array,
+    I64Array, Int256Array, IntervalArray, SerialArray, TimeArray, TimestampArray,
+    TimestampNanosecondArray, TimestamptzArray, Utf8Array,
 };
 use crate::test_utils::rand_array::seed_rand_array_ref;
 use crate::types::DataType;
@@ -38,6 +38,9 @@ pub fn gen_chunk(data_types: &[DataType], size: usize, seed: u64, null_ratio: f6
             DataType::Time => seed_rand_array_ref::<TimeArray>(size, seed, null_ratio),
             DataType::Serial => seed_rand_array_ref::<SerialArray>(size, seed, null_ratio),
             DataType::Timestamp => seed_rand_array_ref::<TimestampArray>(size, seed, null_ratio),
+            DataType::TimestampNanosecond => {
+                seed_rand_array_ref::<TimestampNanosecondArray>(size, seed, null_ratio)
+            }
             DataType::Timestamptz => {
                 seed_rand_array_ref::<TimestamptzArray>(size, seed, null_ratio)
             }