diff --git a/e2e_test/batch/types/timestamp_ns.slt.part b/e2e_test/batch/types/timestamp_ns.slt.part
index 6eb727f678380..e0c969496c256 100644
--- a/e2e_test/batch/types/timestamp_ns.slt.part
+++ b/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 timestamp_ns) 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
 
diff --git a/proto/data.proto b/proto/data.proto
index 9bb15ebcc8d62..478fc3e522c8f 100644
--- a/proto/data.proto
+++ b/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 {
diff --git a/proto/expr.proto b/proto/expr.proto
index 5bd2809f7d314..6edc2db1f747e 100644
--- a/proto/expr.proto
+++ b/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;
diff --git a/src/batch/executors/src/executor/postgres_query.rs b/src/batch/executors/src/executor/postgres_query.rs
index e7ce698e1e841..9a265a195f054 100644
--- a/src/batch/executors/src/executor/postgres_query.rs
+++ b/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
diff --git a/src/common/src/array/arrow/arrow_iceberg.rs b/src/common/src/array/arrow/arrow_iceberg.rs
index 6882ec6f63224..10f0041e24eaa 100644
--- a/src/common/src/array/arrow/arrow_iceberg.rs
+++ b/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(),
diff --git a/src/common/src/array/arrow/arrow_impl.rs b/src/common/src/array/arrow/arrow_impl.rs
index edf83f0a6f54c..01d8bab6489e2 100644
--- a/src/common/src/array/arrow/arrow_impl.rs
+++ b/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 {
@@ -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;
diff --git a/src/common/src/array/chrono_array.rs b/src/common/src/array/chrono_array.rs
index 0e6c9f523d9f7..91fb84d4feb0b 100644
--- a/src/common/src/array/chrono_array.rs
+++ b/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)]
diff --git a/src/common/src/array/mod.rs b/src/common/src/array/mod.rs
index 91e29c429044c..51e823ac8a5e7 100644
--- a/src/common/src/array/mod.rs
+++ b/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;
diff --git a/src/common/src/array/primitive_array.rs b/src/common/src/array/primitive_array.rs
index cdbfa6d95a56b..cfbe539652785 100644
--- a/src/common/src/array/primitive_array.rs
+++ b/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 }
 }
 
diff --git a/src/common/src/array/proto_reader.rs b/src/common/src/array/proto_reader.rs
index 5fbd79d7ab3a6..19b142cf606a2 100644
--- a/src/common/src/array/proto_reader.rs
+++ b/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)?,
diff --git a/src/common/src/hash/key.rs b/src/common/src/hash/key.rs
index 327fec9c5ed13..d8d9e29b319dc 100644
--- a/src/common/src/hash/key.rs
+++ b/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());
diff --git a/src/common/src/row/owned_row.rs b/src/common/src/row/owned_row.rs
index f26608f8948df..88c4fc22501e6 100644
--- a/src/common/src/row/owned_row.rs
+++ b/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!(),
diff --git a/src/common/src/test_utils/rand_array.rs b/src/common/src/test_utils/rand_array.rs
index 9b84d6724ce54..6228379ca7f4d 100644
--- a/src/common/src/test_utils/rand_array.rs
+++ b/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())
diff --git a/src/common/src/test_utils/rand_chunk.rs b/src/common/src/test_utils/rand_chunk.rs
index 9c34335937913..ab01f182df436 100644
--- a/src/common/src/test_utils/rand_chunk.rs
+++ b/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)
             }
diff --git a/src/common/src/types/datetime.rs b/src/common/src/types/datetime.rs
index cf527184a0ddb..d6c77c616408c 100644
--- a/src/common/src/types/datetime.rs
+++ b/src/common/src/types/datetime.rs
@@ -40,8 +40,8 @@ const UNIX_EPOCH_DAYS: i32 = 719_163;
 const LEAP_DAYS: &[i32] = &[0, 31, 29, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31];
 const NORMAL_DAYS: &[i32] = &[0, 31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31];
 
-macro_rules! impl_chrono_wrapper {
-    ($variant_name:ident, $chrono:ty, $pg_type:ident) => {
+macro_rules! impl_chrono_wrapper_without_pgsql_type {
+    ($variant_name:ident, $chrono:ty) => {
         #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
         #[repr(transparent)]
         pub struct $variant_name(pub $chrono);
@@ -67,6 +67,12 @@ macro_rules! impl_chrono_wrapper {
         }
 
         impl ZeroHeapSize for $variant_name {}
+    };
+}
+
+macro_rules! impl_chrono_wrapper {
+    ($variant_name:ident, $chrono:ty, $pg_type:ident) => {
+        impl_chrono_wrapper_without_pgsql_type!($variant_name, $chrono);
 
         impl ToSql for $variant_name {
             accepts!($pg_type);
@@ -100,6 +106,7 @@ macro_rules! impl_chrono_wrapper {
         }
     };
 }
+impl_chrono_wrapper_without_pgsql_type!(TimestampNanosecond, NaiveDateTime);
 
 impl_chrono_wrapper!(Date, NaiveDate, DATE);
 impl_chrono_wrapper!(Timestamp, NaiveDateTime, TIMESTAMP);
@@ -171,15 +178,59 @@ impl FromStr for Timestamp {
         let dt = s
             .parse::<jiff::civil::DateTime>()
             .map_err(|_| ErrorKind::ParseTimestamp)?;
-        Ok(
+        let subsec_nanosecond = dt.subsec_nanosecond();
+        if subsec_nanosecond % 1_000 != 0 {
+            return Err(ErrorKind::ParseTimestamp.into());
+        }
+        Ok(Self::new(
             Date::from_ymd_uncheck(dt.year() as i32, dt.month() as u32, dt.day() as u32)
-                .and_hms_nano_uncheck(
-                    dt.hour() as u32,
-                    dt.minute() as u32,
-                    dt.second() as u32,
-                    dt.subsec_nanosecond() as u32,
+                .0
+                .and_time(
+                    Time::from_hms_micro_uncheck(
+                        dt.hour() as u32,
+                        dt.minute() as u32,
+                        dt.second() as u32,
+                        subsec_nanosecond as u32 / 1_000,
+                    )
+                    .0,
                 ),
-        )
+        ))
+    }
+}
+
+/// Parse a timestampns from varchar.
+///
+/// # Example
+/// ```
+/// use std::str::FromStr;
+///
+/// use risingwave_common::types::TimestampNanosecond;
+///
+/// TimestampNanosecond::from_str("1999-01-08 04:02").unwrap();
+/// TimestampNanosecond::from_str("1999-01-08 04:05:06").unwrap();
+/// TimestampNanosecond::from_str("1999-01-08T04:05:06").unwrap();
+/// TimestampNanosecond::from_str("1999-01-08T04:05:06.123456789").unwrap();
+/// ```
+impl FromStr for TimestampNanosecond {
+    type Err = InvalidParamsError;
+
+    fn from_str(s: &str) -> Result<Self> {
+        let dt = s
+            .parse::<jiff::civil::DateTime>()
+            .map_err(|_| ErrorKind::ParseTimestampNanosecond)?;
+        Ok(Self::new(
+            Date::from_ymd_uncheck(dt.year() as i32, dt.month() as u32, dt.day() as u32)
+                .0
+                .and_time(
+                    Time::from_hms_nano_uncheck(
+                        dt.hour() as u32,
+                        dt.minute() as u32,
+                        dt.second() as u32,
+                        dt.subsec_nanosecond() as u32,
+                    )
+                    .0,
+                ),
+        ))
     }
 }
 
@@ -201,6 +252,22 @@ impl From<Timestamp> for Date {
     }
 }
 
+/// # Example
+/// ```
+/// use std::str::FromStr;
+///
+/// use risingwave_common::types::{Date, TimestampNanosecond};
+///
+/// let ts = TimestampNanosecond::from_str("1999-01-08 04:02").unwrap();
+/// let date = Date::from(ts);
+/// assert_eq!(date, Date::from_str("1999-01-08").unwrap());
+/// ```
+impl From<TimestampNanosecond> for Date {
+    fn from(ts: TimestampNanosecond) -> Self {
+        Date::new(ts.0.date())
+    }
+}
+
 /// In `PostgreSQL`, casting from timestamp to time discards the date part.
 ///
 /// # Example
@@ -219,6 +286,22 @@ impl From<Timestamp> for Time {
     }
 }
 
+/// # Example
+/// ```
+/// use std::str::FromStr;
+///
+/// use risingwave_common::types::{Time, TimestampNanosecond};
+///
+/// let ts = TimestampNanosecond::from_str("1999-01-08 04:02").unwrap();
+/// let time = Time::from(ts);
+/// assert_eq!(time, Time::from_str("04:02").unwrap());
+/// ```
+impl From<TimestampNanosecond> for Time {
+    fn from(ts: TimestampNanosecond) -> Self {
+        Time::new(ts.0.time())
+    }
+}
+
 /// In `PostgreSQL`, casting from interval to time discards the days part.
 ///
 /// # Example
@@ -256,8 +339,10 @@ enum ErrorKind {
     ParseDate,
     #[error("Can't cast string to time (expected format is HH:MM:SS[.D+{{up to 6 digits}}][Z] or HH:MM)")]
     ParseTime,
-    #[error("Can't cast string to timestamp (expected format is YYYY-MM-DD HH:MM:SS[.D+{{up to 9 digits}}] or YYYY-MM-DD HH:MM or YYYY-MM-DD or ISO 8601 format)")]
+    #[error("Can't cast string to timestamp (expected format is YYYY-MM-DD HH:MM:SS[.D+{{up to 6 digits}}] or YYYY-MM-DD HH:MM or YYYY-MM-DD or ISO 8601 format)")]
     ParseTimestamp,
+    #[error("Can't cast string to timestampns (expected format is YYYY-MM-DD HH:MM:SS[.D+{{up to 9 digits}}] or YYYY-MM-DD HH:MM or YYYY-MM-DD or ISO 8601 format)")]
+    ParseTimestampNanosecond,
 }
 
 #[derive(Debug, Error)]
@@ -332,29 +417,6 @@ impl ToText for Time {
     }
 }
 
-impl ToText for Timestamp {
-    fn write<W: std::fmt::Write>(&self, f: &mut W) -> std::fmt::Result {
-        let (ce, year) = self.0.year_ce();
-        let suffix = if ce { "" } else { " BC" };
-        write!(
-            f,
-            "{:04}-{:02}-{:02} {}{}",
-            year,
-            self.0.month(),
-            self.0.day(),
-            self.0.time(),
-            suffix
-        )
-    }
-
-    fn write_with_type<W: std::fmt::Write>(&self, ty: &DataType, f: &mut W) -> std::fmt::Result {
-        match ty {
-            super::DataType::Timestamp => self.write(f),
-            _ => unreachable!(),
-        }
-    }
-}
-
 impl Date {
     pub fn with_days_since_ce(days: i32) -> Result<Self> {
         Ok(Date::new(
@@ -402,21 +464,7 @@ impl Date {
     }
 
     pub fn and_hms_uncheck(self, hour: u32, min: u32, sec: u32) -> Timestamp {
-        self.and_hms_micro_uncheck(hour, min, sec, 0)
-    }
-
-    pub fn and_hms_micro_uncheck(self, hour: u32, min: u32, sec: u32, micro: u32) -> Timestamp {
-        Timestamp::new(
-            self.0
-                .and_time(Time::from_hms_micro_uncheck(hour, min, sec, micro).0),
-        )
-    }
-
-    pub fn and_hms_nano_uncheck(self, hour: u32, min: u32, sec: u32, nano: u32) -> Timestamp {
-        Timestamp::new(
-            self.0
-                .and_time(Time::from_hms_nano_uncheck(hour, min, sec, nano).0),
-        )
+        Timestamp::new(self.0.and_time(Time::from_hms_uncheck(hour, min, sec).0))
     }
 }
 
@@ -481,365 +529,394 @@ impl Time {
     }
 }
 
-// The first 64 bits of protobuf encoding for `Timestamp` type has 2 possible meanings.
-// * When the highest 2 bits are `11` or `00` (i.e. values ranging from `0b1100...00` to `0b0011..11`),
-//   it is *microseconds* since 1970-01-01 midnight. 2^62 microseconds covers 146235 years.
-// * When the highest 2 bits are `10` or `01`, we flip the second bit to get values from `0b1100...00` to `0b0011..11` again.
-//   It is *seconds* since 1970-01-01 midnight. It is then followed by another 32 bits as nanoseconds within a second.
-// Since timestamp is negative when it is less than 1970-1-1, you need to take both cases into account(`11+00`` or `01+10``).
-enum FirstI64 {
-    V0 { usecs: i64 },
-    V1 { secs: i64 },
-}
-impl FirstI64 {
-    pub fn to_protobuf(&self) -> i64 {
-        match self {
-            FirstI64::V0 { usecs } => *usecs,
-            FirstI64::V1 { secs } => secs ^ (0b01 << 62),
-        }
-    }
-
-    pub fn from_protobuf(cur: &mut Cursor<&[u8]>) -> ArrayResult<FirstI64> {
-        let value = cur
-            .read_i64::<BigEndian>()
-            .context("failed to read i64 from Time buffer")?;
-        if Self::is_v1_format_state(value) {
-            let secs = value ^ (0b01 << 62);
-            Ok(FirstI64::V1 { secs })
-        } else {
-            Ok(FirstI64::V0 { usecs: value })
-        }
+/// return the days of the `year-month`
+fn get_mouth_days(year: i32, month: usize) -> i32 {
+    if is_leap_year(year) {
+        LEAP_DAYS[month]
+    } else {
+        NORMAL_DAYS[month]
     }
+}
 
-    fn is_v1_format_state(value: i64) -> bool {
-        let state = (value >> 62) & 0b11;
-        state == 0b10 || state == 0b01
-    }
+fn is_leap_year(year: i32) -> bool {
+    year % 4 == 0 && (year % 100 != 0 || year % 400 == 0)
 }
 
-impl Timestamp {
-    pub fn with_secs_nsecs(secs: i64, nsecs: u32) -> Result<Self> {
-        Ok(Timestamp::new({
-            DateTime::from_timestamp(secs, nsecs)
-                .map(|t| t.naive_utc())
-                .ok_or_else(|| InvalidParamsError::datetime(secs, nsecs))?
-        }))
-    }
+macro_rules! impl_timestamp {
+    ($type:ty) => {
+        impl From<Date> for $type {
+            fn from(date: Date) -> Self {
+                Self::new(date.0.and_time(Time::from_hms_nano_uncheck(0, 0, 0, 0).0))
+            }
+        }
 
-    pub fn from_protobuf(cur: &mut Cursor<&[u8]>) -> ArrayResult<Timestamp> {
-        match FirstI64::from_protobuf(cur)? {
-            FirstI64::V0 { usecs } => Ok(Timestamp::with_micros(usecs)?),
-            FirstI64::V1 { secs } => {
-                let nsecs = cur
-                    .read_u32::<BigEndian>()
-                    .context("failed to read u32 from Time buffer")?;
-                Ok(Timestamp::with_secs_nsecs(secs, nsecs)?)
+        impl CheckedAdd<Interval> for $type {
+            type Output = $type;
+
+            fn checked_add(self, rhs: Interval) -> Option<$type> {
+                let mut date = self.0.date();
+                if rhs.months() != 0 {
+                    // NaiveDate don't support add months. We need calculate manually
+                    let mut day = date.day() as i32;
+                    let mut month = date.month() as i32;
+                    let mut year = date.year();
+                    // Calculate the number of year in this interval
+                    let interval_months = rhs.months();
+                    let year_diff = interval_months / 12;
+                    year += year_diff;
+
+                    // Calculate the number of month in this interval except the added year
+                    // The range of month_diff is (-12, 12) (The month is negative when the interval is
+                    // negative)
+                    let month_diff = interval_months - year_diff * 12;
+                    // The range of new month is (-12, 24) ( original month:[1, 12] + month_diff:(-12, 12) )
+                    month += month_diff;
+                    // Process the overflow months
+                    if month > 12 {
+                        year += 1;
+                        month -= 12;
+                    } else if month <= 0 {
+                        year -= 1;
+                        month += 12;
+                    }
+
+                    // Fix the days after changing date.
+                    // For example, 1970.1.31 + 1 month = 1970.2.28
+                    day = day.min(get_mouth_days(year, month as usize));
+                    date = NaiveDate::from_ymd_opt(year, month as u32, day as u32)?;
+                }
+                let mut datetime = NaiveDateTime::new(date, self.0.time());
+                datetime = datetime.checked_add_signed(Duration::days(rhs.days().into()))?;
+                datetime = datetime.checked_add_signed(Duration::microseconds(rhs.usecs()))?;
+
+                Some(Self::new(datetime))
             }
         }
-    }
 
-    // Since timestamp secs is much smaller than i64, we use the highest 2 bit to store the format information, which is compatible with the old format.
-    // New format: secs(i64) + nsecs(u32)
-    // Old format: micros(i64)
-    pub fn to_protobuf<T: Write>(self, output: &mut T) -> ArrayResult<usize> {
-        let timestamp_size = output
-            .write(
-                &(FirstI64::V1 {
-                    secs: self.0.and_utc().timestamp(),
+        impl ToText for $type {
+            fn write<W: std::fmt::Write>(&self, f: &mut W) -> std::fmt::Result {
+                let (ce, year) = self.0.year_ce();
+                let suffix = if ce { "" } else { " BC" };
+                write!(
+                    f,
+                    "{:04}-{:02}-{:02} {}{}",
+                    year,
+                    self.0.month(),
+                    self.0.day(),
+                    self.0.time(),
+                    suffix
+                )
+            }
+
+            fn write_with_type<W: std::fmt::Write>(
+                &self,
+                ty: &DataType,
+                f: &mut W,
+            ) -> std::fmt::Result {
+                match ty {
+                    super::DataType::Timestamp => self.write(f),
+                    super::DataType::TimestampNanosecond => self.write(f),
+                    _ => unreachable!(),
                 }
-                .to_protobuf())
-                .to_be_bytes(),
-            )
-            .map_err(Into::<ArrayError>::into)?;
-        let timestamp_subsec_nanos_size = output
-            .write(&(self.0.and_utc().timestamp_subsec_nanos()).to_be_bytes())
-            .map_err(Into::<ArrayError>::into)?;
-        Ok(timestamp_subsec_nanos_size + timestamp_size)
-    }
+            }
+        }
 
-    pub fn get_timestamp_nanos(&self) -> i64 {
-        self.0.and_utc().timestamp_nanos_opt().unwrap()
-    }
+        impl $type {
+            pub fn with_secs_nsecs(secs: i64, nsecs: u32) -> Result<Self> {
+                Ok(Self::new({
+                    DateTime::from_timestamp(secs, nsecs)
+                        .map(|t| t.naive_utc())
+                        .ok_or_else(|| InvalidParamsError::datetime(secs, nsecs))?
+                }))
+            }
 
-    pub fn with_millis(timestamp_millis: i64) -> Result<Self> {
-        let secs = timestamp_millis.div_euclid(1_000);
-        let nsecs = timestamp_millis.rem_euclid(1_000) * 1_000_000;
-        Self::with_secs_nsecs(secs, nsecs as u32)
-    }
+            pub fn with_millis(timestamp_millis: i64) -> Result<Self> {
+                let secs = timestamp_millis.div_euclid(1_000);
+                let nsecs = timestamp_millis.rem_euclid(1_000) * 1_000_000;
+                Self::with_secs_nsecs(secs, nsecs as u32)
+            }
 
-    pub fn with_micros(timestamp_micros: i64) -> Result<Self> {
-        let secs = timestamp_micros.div_euclid(1_000_000);
-        let nsecs = timestamp_micros.rem_euclid(1_000_000) * 1000;
-        Self::with_secs_nsecs(secs, nsecs as u32)
-    }
+            pub fn with_micros(timestamp_micros: i64) -> Result<Self> {
+                let secs = timestamp_micros.div_euclid(1_000_000);
+                let nsecs = timestamp_micros.rem_euclid(1_000_000) * 1000;
+                Self::with_secs_nsecs(secs, nsecs as u32)
+            }
 
-    pub fn from_timestamp_uncheck(secs: i64, nsecs: u32) -> Self {
-        Self::new(DateTime::from_timestamp(secs, nsecs).unwrap().naive_utc())
-    }
+            pub fn from_timestamp_uncheck(secs: i64, nsecs: u32) -> Self {
+                Self::new(DateTime::from_timestamp(secs, nsecs).unwrap().naive_utc())
+            }
 
-    /// Truncate the timestamp to the precision of microseconds.
-    ///
-    /// # Example
-    /// ```
-    /// # use risingwave_common::types::Timestamp;
-    /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
-    /// assert_eq!(
-    ///     Timestamp::new(ts).truncate_micros().to_string(),
-    ///     "2001-05-16 20:38:40.123456"
-    /// );
-    /// ```
-    pub fn truncate_micros(self) -> Self {
-        Self::new(
-            self.0
-                .with_nanosecond(self.0.nanosecond() / 1000 * 1000)
-                .unwrap(),
-        )
-    }
+            /// Truncate the timestamp to the precision of microseconds.
+            ///
+            /// # Example
+            /// ```
+            /// # use risingwave_common::types::Timestamp;
+            /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
+            /// assert_eq!(
+            ///     Timestamp::new(ts).truncate_micros().to_string(),
+            ///     "2001-05-16 20:38:40.123456"
+            /// );
+            /// ```
+            pub fn truncate_micros(self) -> Self {
+                Self::new(
+                    self.0
+                        .with_nanosecond(self.0.nanosecond() / 1000 * 1000)
+                        .unwrap(),
+                )
+            }
 
-    /// Truncate the timestamp to the precision of milliseconds.
-    ///
-    /// # Example
-    /// ```
-    /// # use risingwave_common::types::Timestamp;
-    /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
-    /// assert_eq!(
-    ///     Timestamp::new(ts).truncate_millis().to_string(),
-    ///     "2001-05-16 20:38:40.123"
-    /// );
-    /// ```
-    pub fn truncate_millis(self) -> Self {
-        Self::new(
-            self.0
-                .with_nanosecond(self.0.nanosecond() / 1_000_000 * 1_000_000)
-                .unwrap(),
-        )
-    }
+            /// Truncate the timestamp to the precision of milliseconds.
+            ///
+            /// # Example
+            /// ```
+            /// # use risingwave_common::types::Timestamp;
+            /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
+            /// assert_eq!(
+            ///     Timestamp::new(ts).truncate_millis().to_string(),
+            ///     "2001-05-16 20:38:40.123"
+            /// );
+            /// ```
+            pub fn truncate_millis(self) -> Self {
+                Self::new(
+                    self.0
+                        .with_nanosecond(self.0.nanosecond() / 1_000_000 * 1_000_000)
+                        .unwrap(),
+                )
+            }
 
-    /// Truncate the timestamp to the precision of seconds.
-    ///
-    /// # Example
-    /// ```
-    /// # use risingwave_common::types::Timestamp;
-    /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
-    /// assert_eq!(
-    ///     Timestamp::new(ts).truncate_second().to_string(),
-    ///     "2001-05-16 20:38:40"
-    /// );
-    /// ```
-    pub fn truncate_second(self) -> Self {
-        Self::new(self.0.with_nanosecond(0).unwrap())
-    }
+            /// Truncate the timestamp to the precision of seconds.
+            ///
+            /// # Example
+            /// ```
+            /// # use risingwave_common::types::Timestamp;
+            /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
+            /// assert_eq!(
+            ///     Timestamp::new(ts).truncate_second().to_string(),
+            ///     "2001-05-16 20:38:40"
+            /// );
+            /// ```
+            pub fn truncate_second(self) -> Self {
+                Self::new(self.0.with_nanosecond(0).unwrap())
+            }
 
-    /// Truncate the timestamp to the precision of minutes.
-    ///
-    /// # Example
-    /// ```
-    /// # use risingwave_common::types::Timestamp;
-    /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
-    /// assert_eq!(
-    ///     Timestamp::new(ts).truncate_minute().to_string(),
-    ///     "2001-05-16 20:38:00"
-    /// );
-    /// ```
-    pub fn truncate_minute(self) -> Self {
-        Date::new(self.0.date()).and_hms_uncheck(self.0.hour(), self.0.minute(), 0)
-    }
+            /// Truncate the timestamp to the precision of minutes.
+            ///
+            /// # Example
+            /// ```
+            /// # use risingwave_common::types::Timestamp;
+            /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
+            /// assert_eq!(
+            ///     Timestamp::new(ts).truncate_minute().to_string(),
+            ///     "2001-05-16 20:38:00"
+            /// );
+            /// ```
+            pub fn truncate_minute(self) -> Self {
+                Self::new(
+                    Date::new(self.0.date()).0.and_time(
+                        Time::from_hms_nano_uncheck(self.0.hour(), self.0.minute(), 0, 0).0,
+                    ),
+                )
+            }
 
-    /// Truncate the timestamp to the precision of hours.
-    ///
-    /// # Example
-    /// ```
-    /// # use risingwave_common::types::Timestamp;
-    /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
-    /// assert_eq!(
-    ///     Timestamp::new(ts).truncate_hour().to_string(),
-    ///     "2001-05-16 20:00:00"
-    /// );
-    /// ```
-    pub fn truncate_hour(self) -> Self {
-        Date::new(self.0.date()).and_hms_uncheck(self.0.hour(), 0, 0)
-    }
+            /// Truncate the timestamp to the precision of hours.
+            ///
+            /// # Example
+            /// ```
+            /// # use risingwave_common::types::Timestamp;
+            /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
+            /// assert_eq!(
+            ///     Timestamp::new(ts).truncate_hour().to_string(),
+            ///     "2001-05-16 20:00:00"
+            /// );
+            /// ```
+            pub fn truncate_hour(self) -> Self {
+                Self::new(
+                    Date::new(self.0.date())
+                        .0
+                        .and_time(Time::from_hms_nano_uncheck(self.0.hour(), 0, 0, 0).0),
+                )
+            }
 
-    /// Truncate the timestamp to the precision of days.
-    ///
-    /// # Example
-    /// ```
-    /// # use risingwave_common::types::Timestamp;
-    /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
-    /// assert_eq!(
-    ///     Timestamp::new(ts).truncate_day().to_string(),
-    ///     "2001-05-16 00:00:00"
-    /// );
-    /// ```
-    pub fn truncate_day(self) -> Self {
-        Date::new(self.0.date()).into()
-    }
+            /// Truncate the timestamp to the precision of days.
+            ///
+            /// # Example
+            /// ```
+            /// # use risingwave_common::types::Timestamp;
+            /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
+            /// assert_eq!(
+            ///     Timestamp::new(ts).truncate_day().to_string(),
+            ///     "2001-05-16 00:00:00"
+            /// );
+            /// ```
+            pub fn truncate_day(self) -> Self {
+                Date::new(self.0.date()).into()
+            }
 
-    /// Truncate the timestamp to the precision of weeks.
-    ///
-    /// # Example
-    /// ```
-    /// # use risingwave_common::types::Timestamp;
-    /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
-    /// assert_eq!(
-    ///     Timestamp::new(ts).truncate_week().to_string(),
-    ///     "2001-05-14 00:00:00"
-    /// );
-    /// ```
-    pub fn truncate_week(self) -> Self {
-        Date::new(self.0.date().week(Weekday::Mon).first_day()).into()
-    }
+            /// Truncate the timestamp to the precision of weeks.
+            ///
+            /// # Example
+            /// ```
+            /// # use risingwave_common::types::Timestamp;
+            /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
+            /// assert_eq!(
+            ///     Timestamp::new(ts).truncate_week().to_string(),
+            ///     "2001-05-14 00:00:00"
+            /// );
+            /// ```
+            pub fn truncate_week(self) -> Self {
+                Date::new(self.0.date().week(Weekday::Mon).first_day()).into()
+            }
 
-    /// Truncate the timestamp to the precision of months.
-    ///
-    /// # Example
-    /// ```
-    /// # use risingwave_common::types::Timestamp;
-    /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
-    /// assert_eq!(
-    ///     Timestamp::new(ts).truncate_month().to_string(),
-    ///     "2001-05-01 00:00:00"
-    /// );
-    /// ```
-    pub fn truncate_month(self) -> Self {
-        Date::new(self.0.date().with_day(1).unwrap()).into()
-    }
+            /// Truncate the timestamp to the precision of months.
+            ///
+            /// # Example
+            /// ```
+            /// # use risingwave_common::types::Timestamp;
+            /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
+            /// assert_eq!(
+            ///     Timestamp::new(ts).truncate_month().to_string(),
+            ///     "2001-05-01 00:00:00"
+            /// );
+            /// ```
+            pub fn truncate_month(self) -> Self {
+                Date::new(self.0.date().with_day(1).unwrap()).into()
+            }
 
-    /// Truncate the timestamp to the precision of quarters.
-    ///
-    /// # Example
-    /// ```
-    /// # use risingwave_common::types::Timestamp;
-    /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
-    /// assert_eq!(
-    ///     Timestamp::new(ts).truncate_quarter().to_string(),
-    ///     "2001-04-01 00:00:00"
-    /// );
-    /// ```
-    pub fn truncate_quarter(self) -> Self {
-        Date::from_ymd_uncheck(self.0.year(), self.0.month0() / 3 * 3 + 1, 1).into()
-    }
+            /// Truncate the timestamp to the precision of quarters.
+            ///
+            /// # Example
+            /// ```
+            /// # use risingwave_common::types::Timestamp;
+            /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
+            /// assert_eq!(
+            ///     Timestamp::new(ts).truncate_quarter().to_string(),
+            ///     "2001-04-01 00:00:00"
+            /// );
+            /// ```
+            pub fn truncate_quarter(self) -> Self {
+                Date::from_ymd_uncheck(self.0.year(), self.0.month0() / 3 * 3 + 1, 1).into()
+            }
 
-    /// Truncate the timestamp to the precision of years.
-    ///
-    /// # Example
-    /// ```
-    /// # use risingwave_common::types::Timestamp;
-    /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
-    /// assert_eq!(
-    ///     Timestamp::new(ts).truncate_year().to_string(),
-    ///     "2001-01-01 00:00:00"
-    /// );
-    /// ```
-    pub fn truncate_year(self) -> Self {
-        Date::from_ymd_uncheck(self.0.year(), 1, 1).into()
-    }
+            /// Truncate the timestamp to the precision of years.
+            ///
+            /// # Example
+            /// ```
+            /// # use risingwave_common::types::Timestamp;
+            /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
+            /// assert_eq!(
+            ///     Timestamp::new(ts).truncate_year().to_string(),
+            ///     "2001-01-01 00:00:00"
+            /// );
+            /// ```
+            pub fn truncate_year(self) -> Self {
+                Date::from_ymd_uncheck(self.0.year(), 1, 1).into()
+            }
 
-    /// Truncate the timestamp to the precision of decades.
-    ///
-    /// # Example
-    /// ```
-    /// # use risingwave_common::types::Timestamp;
-    /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
-    /// assert_eq!(
-    ///     Timestamp::new(ts).truncate_decade().to_string(),
-    ///     "2000-01-01 00:00:00"
-    /// );
-    /// ```
-    pub fn truncate_decade(self) -> Self {
-        Date::from_ymd_uncheck(self.0.year() / 10 * 10, 1, 1).into()
-    }
+            /// Truncate the timestamp to the precision of decades.
+            ///
+            /// # Example
+            /// ```
+            /// # use risingwave_common::types::Timestamp;
+            /// let ts = "2001-05-16T20:38:40.123456789".parse().unwrap();
+            /// assert_eq!(
+            ///     Timestamp::new(ts).truncate_decade().to_string(),
+            ///     "2000-01-01 00:00:00"
+            /// );
+            /// ```
+            pub fn truncate_decade(self) -> Self {
+                Date::from_ymd_uncheck(self.0.year() / 10 * 10, 1, 1).into()
+            }
 
-    /// Truncate the timestamp to the precision of centuries.
-    ///
-    /// # Example
-    /// ```
-    /// # use risingwave_common::types::Timestamp;
-    /// let ts = "3202-05-16T20:38:40.123456789".parse().unwrap();
-    /// assert_eq!(
-    ///     Timestamp::new(ts).truncate_century().to_string(),
-    ///     "3201-01-01 00:00:00"
-    /// );
-    /// ```
-    pub fn truncate_century(self) -> Self {
-        Date::from_ymd_uncheck((self.0.year() - 1) / 100 * 100 + 1, 1, 1).into()
-    }
+            /// Truncate the timestamp to the precision of centuries.
+            ///
+            /// # Example
+            /// ```
+            /// # use risingwave_common::types::Timestamp;
+            /// let ts = "3202-05-16T20:38:40.123456789".parse().unwrap();
+            /// assert_eq!(
+            ///     Timestamp::new(ts).truncate_century().to_string(),
+            ///     "3201-01-01 00:00:00"
+            /// );
+            /// ```
+            pub fn truncate_century(self) -> Self {
+                Date::from_ymd_uncheck((self.0.year() - 1) / 100 * 100 + 1, 1, 1).into()
+            }
 
-    /// Truncate the timestamp to the precision of millenniums.
-    ///
-    /// # Example
-    /// ```
-    /// # use risingwave_common::types::Timestamp;
-    /// let ts = "3202-05-16T20:38:40.123456789".parse().unwrap();
-    /// assert_eq!(
-    ///     Timestamp::new(ts).truncate_millennium().to_string(),
-    ///     "3001-01-01 00:00:00"
-    /// );
-    /// ```
-    pub fn truncate_millennium(self) -> Self {
-        Date::from_ymd_uncheck((self.0.year() - 1) / 1000 * 1000 + 1, 1, 1).into()
-    }
+            /// Truncate the timestamp to the precision of millenniums.
+            ///
+            /// # Example
+            /// ```
+            /// # use risingwave_common::types::Timestamp;
+            /// let ts = "3202-05-16T20:38:40.123456789".parse().unwrap();
+            /// assert_eq!(
+            ///     Timestamp::new(ts).truncate_millennium().to_string(),
+            ///     "3001-01-01 00:00:00"
+            /// );
+            /// ```
+            pub fn truncate_millennium(self) -> Self {
+                Date::from_ymd_uncheck((self.0.year() - 1) / 1000 * 1000 + 1, 1, 1).into()
+            }
+        }
+    };
 }
+impl_timestamp!(Timestamp);
+impl_timestamp!(TimestampNanosecond);
 
-impl From<Date> for Timestamp {
-    fn from(date: Date) -> Self {
-        date.and_hms_uncheck(0, 0, 0)
+impl From<Timestamp> for TimestampNanosecond {
+    fn from(ts: Timestamp) -> Self {
+        TimestampNanosecond::new(ts.0)
     }
 }
 
-/// return the days of the `year-month`
-fn get_mouth_days(year: i32, month: usize) -> i32 {
-    if is_leap_year(year) {
-        LEAP_DAYS[month]
-    } else {
-        NORMAL_DAYS[month]
+impl From<TimestampNanosecond> for Timestamp {
+    fn from(ts: TimestampNanosecond) -> Self {
+        let ts = ts.truncate_micros();
+        Timestamp::new(ts.0)
     }
 }
 
-fn is_leap_year(year: i32) -> bool {
-    year % 4 == 0 && (year % 100 != 0 || year % 400 == 0)
+impl TimestampNanosecond {
+    pub fn from_protobuf(cur: &mut Cursor<&[u8]>) -> ArrayResult<TimestampNanosecond> {
+        let secs = cur
+            .read_i64::<BigEndian>()
+            .context("failed to read i64 from TimestampNanosecond buffer")?;
+        let nsecs = cur
+            .read_u32::<BigEndian>()
+            .context("failed to read u32 from TimestampNanosecond buffer")?;
+        Ok(TimestampNanosecond::with_secs_nsecs(secs, nsecs)?)
+    }
+
+    pub fn to_protobuf<T: Write>(self, output: &mut T) -> ArrayResult<usize> {
+        let timestamp_size = output
+            .write(&(self.0.and_utc().timestamp()).to_be_bytes())
+            .map_err(Into::<ArrayError>::into)?;
+        let timestamp_subsec_nanos_size = output
+            .write(&(self.0.and_utc().timestamp_subsec_nanos()).to_be_bytes())
+            .map_err(Into::<ArrayError>::into)?;
+        Ok(timestamp_subsec_nanos_size + timestamp_size)
+    }
+
+    pub fn get_timestamp_nanos(&self) -> i64 {
+        self.0.and_utc().timestamp_nanos_opt().unwrap()
+    }
 }
 
-impl CheckedAdd<Interval> for Timestamp {
-    type Output = Timestamp;
-
-    fn checked_add(self, rhs: Interval) -> Option<Timestamp> {
-        let mut date = self.0.date();
-        if rhs.months() != 0 {
-            // NaiveDate don't support add months. We need calculate manually
-            let mut day = date.day() as i32;
-            let mut month = date.month() as i32;
-            let mut year = date.year();
-            // Calculate the number of year in this interval
-            let interval_months = rhs.months();
-            let year_diff = interval_months / 12;
-            year += year_diff;
-
-            // Calculate the number of month in this interval except the added year
-            // The range of month_diff is (-12, 12) (The month is negative when the interval is
-            // negative)
-            let month_diff = interval_months - year_diff * 12;
-            // The range of new month is (-12, 24) ( original month:[1, 12] + month_diff:(-12, 12) )
-            month += month_diff;
-            // Process the overflow months
-            if month > 12 {
-                year += 1;
-                month -= 12;
-            } else if month <= 0 {
-                year -= 1;
-                month += 12;
-            }
+impl Timestamp {
+    pub fn from_protobuf(cur: &mut Cursor<&[u8]>) -> ArrayResult<Timestamp> {
+        let micros = cur
+            .read_i64::<BigEndian>()
+            .context("failed to read i64 from Timestamp buffer")?;
+        Ok(Timestamp::with_micros(micros)?)
+    }
 
-            // Fix the days after changing date.
-            // For example, 1970.1.31 + 1 month = 1970.2.28
-            day = day.min(get_mouth_days(year, month as usize));
-            date = NaiveDate::from_ymd_opt(year, month as u32, day as u32)?;
-        }
-        let mut datetime = NaiveDateTime::new(date, self.0.time());
-        datetime = datetime.checked_add_signed(Duration::days(rhs.days().into()))?;
-        datetime = datetime.checked_add_signed(Duration::microseconds(rhs.usecs()))?;
+    pub fn to_protobuf<T: Write>(self, output: &mut T) -> ArrayResult<usize> {
+        output
+            .write(&(self.0.and_utc().timestamp_micros()).to_be_bytes())
+            .map_err(Into::into)
+    }
 
-        Some(Timestamp::new(datetime))
+    pub fn get_timestamp_nanos(&self) -> i64 {
+        self.0.and_utc().timestamp_nanos_opt().unwrap()
     }
 }
 
diff --git a/src/common/src/types/macros.rs b/src/common/src/types/macros.rs
index a35a3fc60278a..323b1b7ab66c1 100644
--- a/src/common/src/types/macros.rs
+++ b/src/common/src/types/macros.rs
@@ -53,6 +53,7 @@ macro_rules! for_all_variants {
             { Date,         Date,         date,         $crate::types::Date,        $crate::types::Date,                $crate::array::DateArray,           $crate::array::DateArrayBuilder         },
             { Time,         Time,         time,         $crate::types::Time,        $crate::types::Time,                $crate::array::TimeArray,           $crate::array::TimeArrayBuilder         },
             { Timestamp,    Timestamp,    timestamp,    $crate::types::Timestamp,   $crate::types::Timestamp,           $crate::array::TimestampArray,      $crate::array::TimestampArrayBuilder    },
+            { TimestampNanosecond,    TimestampNanosecond,    timestampns,    $crate::types::TimestampNanosecond,   $crate::types::TimestampNanosecond,           $crate::array::TimestampNanosecondArray,      $crate::array::TimestampNanosecondArrayBuilder    },
             { Timestamptz,  Timestamptz,  timestamptz,  $crate::types::Timestamptz, $crate::types::Timestamptz,         $crate::array::TimestamptzArray,    $crate::array::TimestamptzArrayBuilder  },
             { Jsonb,        Jsonb,        jsonb,        $crate::types::JsonbVal,    $crate::types::JsonbRef<'scalar>,   $crate::array::JsonbArray,          $crate::array::JsonbArrayBuilder        },
             { Serial,       Serial,       serial,       $crate::types::Serial,      $crate::types::Serial,              $crate::array::SerialArray,         $crate::array::SerialArrayBuilder       },
diff --git a/src/common/src/types/map_type.rs b/src/common/src/types/map_type.rs
index 2268591a52b26..b75921fe0d121 100644
--- a/src/common/src/types/map_type.rs
+++ b/src/common/src/types/map_type.rs
@@ -111,6 +111,7 @@ impl MapType {
             | DataType::Date
             | DataType::Time
             | DataType::Timestamp
+            | DataType::TimestampNanosecond
             | DataType::Timestamptz
             | DataType::Interval
             | DataType::Struct(_)
diff --git a/src/common/src/types/mod.rs b/src/common/src/types/mod.rs
index 790bb0a7f8eaf..97b1a50bae304 100644
--- a/src/common/src/types/mod.rs
+++ b/src/common/src/types/mod.rs
@@ -76,7 +76,7 @@ pub use fields::Fields;
 pub use risingwave_fields_derive::Fields;
 
 pub use self::cow::DatumCow;
-pub use self::datetime::{Date, Time, Timestamp};
+pub use self::datetime::{Date, Time, Timestamp, TimestampNanosecond};
 pub use self::decimal::{Decimal, PowError as DecimalPowError};
 pub use self::interval::{test_utils, DateTimeField, Interval, IntervalDisplay};
 pub use self::jsonb::{JsonbRef, JsonbVal};
@@ -150,6 +150,9 @@ pub enum DataType {
     #[display("timestamp without time zone")]
     #[from_str(regex = "(?i)^timestamp$|^timestamp without time zone$")]
     Timestamp,
+    #[display("timestamp_ns without time zone")]
+    #[from_str(regex = "(?i)^timestamp_ns$|^timestamp_ns without time zone$")]
+    TimestampNanosecond,
     #[display("timestamp with time zone")]
     #[from_str(regex = "(?i)^timestamptz$|^timestamp with time zone$")]
     Timestamptz,
@@ -208,6 +211,7 @@ impl TryFrom<DataTypeName> for DataType {
             DataTypeName::Bytea => Ok(DataType::Bytea),
             DataTypeName::Date => Ok(DataType::Date),
             DataTypeName::Timestamp => Ok(DataType::Timestamp),
+            DataTypeName::TimestampNanosecond => Ok(DataType::TimestampNanosecond),
             DataTypeName::Timestamptz => Ok(DataType::Timestamptz),
             DataTypeName::Time => Ok(DataType::Time),
             DataTypeName::Interval => Ok(DataType::Interval),
@@ -234,6 +238,7 @@ impl From<&PbDataType> for DataType {
             PbTypeName::Date => DataType::Date,
             PbTypeName::Time => DataType::Time,
             PbTypeName::Timestamp => DataType::Timestamp,
+            PbTypeName::TimestampNanosecond => DataType::TimestampNanosecond,
             PbTypeName::Timestamptz => DataType::Timestamptz,
             PbTypeName::Decimal => DataType::Decimal,
             PbTypeName::Interval => DataType::Interval,
@@ -286,6 +291,7 @@ impl From<DataTypeName> for PbTypeName {
             DataTypeName::List => PbTypeName::List,
             DataTypeName::Int256 => PbTypeName::Int256,
             DataTypeName::Map => PbTypeName::Map,
+            DataTypeName::TimestampNanosecond => PbTypeName::TimestampNanosecond,
         }
     }
 }
@@ -313,6 +319,7 @@ pub mod data_types {
             DataType::Date
                 | DataType::Time
                 | DataType::Timestamp
+                | DataType::TimestampNanosecond
                 | DataType::Timestamptz
                 | DataType::Interval
         };
@@ -362,6 +369,7 @@ impl DataType {
             | DataType::Varchar
             | DataType::Time
             | DataType::Timestamp
+            | DataType::TimestampNanosecond
             | DataType::Timestamptz
             | DataType::Interval
             | DataType::Bytea
@@ -406,6 +414,7 @@ impl DataType {
         match input {
             DataType::Timestamptz => Some(DataType::Timestamptz),
             DataType::Timestamp | DataType::Date => Some(DataType::Timestamp),
+            DataType::TimestampNanosecond => Some(DataType::TimestampNanosecond),
             _ => None,
         }
     }
@@ -890,7 +899,10 @@ impl ScalarImpl {
                     .ok_or_else(|| "invalid value of Jsonb".to_owned())?,
             ),
             DataType::Int256 => Self::Int256(Int256::from_binary(bytes)?),
-            DataType::Struct(_) | DataType::List(_) | DataType::Map(_) => {
+            DataType::Struct(_)
+            | DataType::List(_)
+            | DataType::Map(_)
+            | DataType::TimestampNanosecond => {
                 return Err(format!("unsupported data type: {}", data_type).into());
             }
         };
@@ -914,6 +926,7 @@ impl ScalarImpl {
             DataType::Varchar => s.into(),
             DataType::Date => Date::from_str(s)?.into(),
             DataType::Timestamp => Timestamp::from_str(s)?.into(),
+            DataType::TimestampNanosecond => TimestampNanosecond::from_str(s)?.into(),
             // We only handle the case with timezone here, and leave the implicit session timezone case
             // for later phase.
             DataType::Timestamptz => Timestamptz::from_str(s)?.into(),
@@ -1019,6 +1032,10 @@ impl ScalarRefImpl<'_> {
                 v.0.and_utc().timestamp().serialize(&mut *ser)?;
                 v.0.and_utc().timestamp_subsec_nanos().serialize(ser)?;
             }
+            Self::TimestampNanosecond(v) => {
+                v.0.and_utc().timestamp().serialize(&mut *ser)?;
+                v.0.and_utc().timestamp_subsec_nanos().serialize(ser)?;
+            }
             Self::Timestamptz(v) => v.serialize(ser)?,
             Self::Time(v) => {
                 v.0.num_seconds_from_midnight().serialize(&mut *ser)?;
@@ -1074,6 +1091,12 @@ impl ScalarImpl {
                 Timestamp::with_secs_nsecs(secs, nsecs)
                     .map_err(|e| memcomparable::Error::Message(e.to_report_string()))?
             }),
+            Ty::TimestampNanosecond => Self::TimestampNanosecond({
+                let secs = i64::deserialize(&mut *de)?;
+                let nsecs = u32::deserialize(de)?;
+                TimestampNanosecond::with_secs_nsecs(secs, nsecs)
+                    .map_err(|e| memcomparable::Error::Message(e.to_report_string()))?
+            }),
             Ty::Timestamptz => Self::Timestamptz(Timestamptz::deserialize(de)?),
             Ty::Date => Self::Date({
                 let days = i32::deserialize(de)?;
@@ -1238,6 +1261,12 @@ mod tests {
                     ScalarImpl::Timestamp(Timestamp::from_timestamp_uncheck(23333333, 2333)),
                     DataType::Timestamp,
                 ),
+                DataTypeName::TimestampNanosecond => (
+                    ScalarImpl::TimestampNanosecond(TimestampNanosecond::from_timestamp_uncheck(
+                        23333333, 2333,
+                    )),
+                    DataType::TimestampNanosecond,
+                ),
                 DataTypeName::Timestamptz => (
                     ScalarImpl::Timestamptz(Timestamptz::from_micros(233333333)),
                     DataType::Timestamptz,
diff --git a/src/common/src/types/postgres_type.rs b/src/common/src/types/postgres_type.rs
index 89667d2f7995d..beb718cfbe1d3 100644
--- a/src/common/src/types/postgres_type.rs
+++ b/src/common/src/types/postgres_type.rs
@@ -69,6 +69,7 @@ impl DataType {
                     $(
                     DataType::$enum => $len,
                     )*
+                    DataType::TimestampNanosecond => 8,
                     DataType::Serial => 8,
                     DataType::Int256 => -1,
                     DataType::List(_) | DataType::Struct(_) | DataType::Map(_) => -1,
@@ -121,11 +122,13 @@ impl DataType {
                         DataType::Struct(_) => 2287, // pseudo-type of array[struct] (see `pg_type.dat`)
                         DataType::List { .. } => unreachable!("Never reach here!"),
                         DataType::Map(_) => 1304,
+                        DataType::TimestampNanosecond => 1306,
                     }
                     DataType::Serial => 20,
                     // XXX: what does the oid mean here? Why we don't have from_oid for them?
                     DataType::Int256 => 1301,
                     DataType::Map(_) => 1303,
+                    DataType::TimestampNanosecond => 1305,
                     // TODO: Support to give a new oid for custom struct type. #9434
                     DataType::Struct(_) => 2249,  // pseudo-type of struct (see `pg_type.dat`)
                 }
@@ -145,6 +148,7 @@ impl DataType {
                     DataType::List(_) => "list",
                     DataType::Serial => "serial",
                     DataType::Int256 => "rw_int256",
+                    DataType::TimestampNanosecond => "timestamp_nano",
                     DataType::Map(_) => "map",
                 }
             }
diff --git a/src/common/src/types/scalar_impl.rs b/src/common/src/types/scalar_impl.rs
index fc0ec2eae7486..cad9fcc7aae40 100644
--- a/src/common/src/types/scalar_impl.rs
+++ b/src/common/src/types/scalar_impl.rs
@@ -245,6 +245,28 @@ impl ScalarRef<'_> for Timestamp {
     }
 }
 
+/// Implement `Scalar` for `Timestamp`.
+impl Scalar for TimestampNanosecond {
+    type ScalarRefType<'a> = TimestampNanosecond;
+
+    fn as_scalar_ref(&self) -> TimestampNanosecond {
+        *self
+    }
+}
+
+/// Implement `ScalarRef` for `Timestamp`.
+impl ScalarRef<'_> for TimestampNanosecond {
+    type ScalarType = TimestampNanosecond;
+
+    fn to_owned_scalar(&self) -> TimestampNanosecond {
+        *self
+    }
+
+    fn hash_scalar<H: std::hash::Hasher>(&self, state: &mut H) {
+        self.hash(state)
+    }
+}
+
 /// Implement `Scalar` for `Time`.
 impl Scalar for Time {
     type ScalarRefType<'a> = Time;
diff --git a/src/common/src/types/to_binary.rs b/src/common/src/types/to_binary.rs
index d497297a047c2..0b3933e9c8b37 100644
--- a/src/common/src/types/to_binary.rs
+++ b/src/common/src/types/to_binary.rs
@@ -17,8 +17,8 @@ use postgres_types::{ToSql, Type};
 use rw_iter_util::ZipEqFast;
 
 use super::{
-    DataType, Date, Decimal, Interval, ScalarRefImpl, Serial, Time, Timestamp, Timestamptz, F32,
-    F64,
+    DataType, Date, Decimal, Interval, ScalarRefImpl, Serial, Time, Timestamp, TimestampNanosecond,
+    Timestamptz, F32, F64,
 };
 use crate::array::{ListRef, StructRef};
 use crate::error::NotImplemented;
@@ -74,6 +74,7 @@ implement_using_to_sql! {
     { Time, Time, |x: &Time| x.0 },
     { Date, Date, |x: &Date| x.0 },
     { Timestamp, Timestamp, |x: &Timestamp| x.0 },
+    { TimestampNanosecond, TimestampNanosecond, |x: &TimestampNanosecond| x.0 },
     { Decimal, Decimal, |x| x },
     { Interval, Interval, |x| x },
     { Serial, Serial, |x: &Serial| x.0 },
@@ -156,6 +157,7 @@ impl ToBinary for ScalarRefImpl<'_> {
             ScalarRefImpl::Interval(v) => v.to_binary_with_type(ty),
             ScalarRefImpl::Date(v) => v.to_binary_with_type(ty),
             ScalarRefImpl::Timestamp(v) => v.to_binary_with_type(ty),
+            ScalarRefImpl::TimestampNanosecond(v) => v.to_binary_with_type(ty),
             ScalarRefImpl::Timestamptz(v) => v.to_binary_with_type(ty),
             ScalarRefImpl::Time(v) => v.to_binary_with_type(ty),
             ScalarRefImpl::Bytea(v) => v.to_binary_with_type(ty),
diff --git a/src/common/src/types/to_sql.rs b/src/common/src/types/to_sql.rs
index 64db7b933f0ce..c5a5254c34746 100644
--- a/src/common/src/types/to_sql.rs
+++ b/src/common/src/types/to_sql.rs
@@ -44,7 +44,10 @@ impl ToSql for ScalarImpl {
             ScalarImpl::Time(v) => v.to_sql(ty, out),
             ScalarImpl::Bytea(v) => (&**v).to_sql(ty, out),
             ScalarImpl::Jsonb(v) => v.to_sql(ty, out),
-            ScalarImpl::Int256(_) | ScalarImpl::Struct(_) | ScalarImpl::List(_) => {
+            ScalarImpl::Int256(_)
+            | ScalarImpl::Struct(_)
+            | ScalarImpl::List(_)
+            | ScalarImpl::TimestampNanosecond(_) => {
                 bail_not_implemented!("the postgres encoding for {ty} is unsupported")
             }
             ScalarImpl::Map(_) => todo!(),
@@ -84,7 +87,10 @@ impl ToSql for ScalarRefImpl<'_> {
             ScalarRefImpl::Time(v) => v.to_sql(ty, out),
             ScalarRefImpl::Bytea(v) => (&**v).to_sql(ty, out),
             ScalarRefImpl::Jsonb(v) => v.to_sql(ty, out),
-            ScalarRefImpl::Int256(_) | ScalarRefImpl::Struct(_) | ScalarRefImpl::List(_) => {
+            ScalarRefImpl::Int256(_)
+            | ScalarRefImpl::Struct(_)
+            | ScalarRefImpl::List(_)
+            | ScalarRefImpl::TimestampNanosecond(_) => {
                 bail_not_implemented!("the postgres encoding for {ty} is unsupported")
             }
             ScalarRefImpl::Map(_) => todo!(),
diff --git a/src/common/src/types/with_data_type.rs b/src/common/src/types/with_data_type.rs
index f159a4d71a126..17f7207b12758 100644
--- a/src/common/src/types/with_data_type.rs
+++ b/src/common/src/types/with_data_type.rs
@@ -19,7 +19,7 @@ use bytes::Bytes;
 
 use super::{
     DataType, Date, Decimal, Fields, Int256, Interval, JsonbRef, JsonbVal, Serial, StructType,
-    Time, Timestamp, Timestamptz, F32, F64,
+    Time, Timestamp, TimestampNanosecond, Timestamptz, F32, F64,
 };
 
 /// A trait for all physical types that can be associated with a [`DataType`].
@@ -103,6 +103,7 @@ impl_with_data_type!(String, DataType::Varchar);
 impl_with_data_type!(Date, DataType::Date);
 impl_with_data_type!(Time, DataType::Time);
 impl_with_data_type!(Timestamp, DataType::Timestamp);
+impl_with_data_type!(TimestampNanosecond, DataType::TimestampNanosecond);
 impl_with_data_type!(Timestamptz, DataType::Timestamptz);
 impl_with_data_type!(Interval, DataType::Interval);
 impl_with_data_type!(Vec<u8>, DataType::Bytea);
diff --git a/src/common/src/util/memcmp_encoding.rs b/src/common/src/util/memcmp_encoding.rs
index 367a09aec42b2..eb64740285484 100644
--- a/src/common/src/util/memcmp_encoding.rs
+++ b/src/common/src/util/memcmp_encoding.rs
@@ -23,8 +23,8 @@ use super::iter_util::{ZipEqDebug, ZipEqFast};
 use crate::array::{ArrayImpl, DataChunk};
 use crate::row::{OwnedRow, Row};
 use crate::types::{
-    DataType, Date, Datum, Int256, ScalarImpl, Serial, Time, Timestamp, Timestamptz, ToDatumRef,
-    F32, F64,
+    DataType, Date, Datum, Int256, ScalarImpl, Serial, Time, Timestamp, TimestampNanosecond,
+    Timestamptz, ToDatumRef, F32, F64,
 };
 use crate::util::sort_util::{ColumnOrder, OrderType};
 
@@ -143,6 +143,7 @@ fn calculate_encoded_size_inner(
             DataType::Date => size_of::<Date>(),
             DataType::Time => size_of::<Time>(),
             DataType::Timestamp => size_of::<Timestamp>(),
+            DataType::TimestampNanosecond => size_of::<TimestampNanosecond>(),
             DataType::Timestamptz => size_of::<Timestamptz>(),
             DataType::Boolean => size_of::<u8>(),
             // Interval is serialized as (i32, i32, i64)
diff --git a/src/common/src/util/value_encoding/mod.rs b/src/common/src/util/value_encoding/mod.rs
index 40ebafbd4eb15..d995620f18aee 100644
--- a/src/common/src/util/value_encoding/mod.rs
+++ b/src/common/src/util/value_encoding/mod.rs
@@ -220,6 +220,11 @@ fn serialize_scalar(value: ScalarRefImpl<'_>, buf: &mut impl BufMut) {
             v.0.and_utc().timestamp_subsec_nanos(),
             buf,
         ),
+        ScalarRefImpl::TimestampNanosecond(v) => serialize_timestamp(
+            v.0.and_utc().timestamp(),
+            v.0.and_utc().timestamp_subsec_nanos(),
+            buf,
+        ),
         ScalarRefImpl::Timestamptz(v) => buf.put_i64_le(v.timestamp_micros()),
         ScalarRefImpl::Time(v) => {
             serialize_time(v.0.num_seconds_from_midnight(), v.0.nanosecond(), buf)
@@ -247,6 +252,7 @@ fn estimate_serialize_scalar_size(value: ScalarRefImpl<'_>) -> usize {
         ScalarRefImpl::Interval(_) => estimate_serialize_interval_size(),
         ScalarRefImpl::Date(_) => estimate_serialize_date_size(),
         ScalarRefImpl::Timestamp(_) => estimate_serialize_timestamp_size(),
+        ScalarRefImpl::TimestampNanosecond(_) => estimate_serialize_timestamp_size(),
         ScalarRefImpl::Timestamptz(_) => 8,
         ScalarRefImpl::Time(_) => estimate_serialize_time_size(),
         // not exact as we use internal encoding size to estimate the json string size
@@ -346,6 +352,9 @@ fn deserialize_value(ty: &DataType, data: &mut impl Buf) -> Result<ScalarImpl> {
         DataType::Interval => ScalarImpl::Interval(deserialize_interval(data)?),
         DataType::Time => ScalarImpl::Time(deserialize_time(data)?),
         DataType::Timestamp => ScalarImpl::Timestamp(deserialize_timestamp(data)?),
+        DataType::TimestampNanosecond => {
+            ScalarImpl::TimestampNanosecond(deserialize_timestamp_nanosecond(data)?)
+        }
         DataType::Timestamptz => {
             ScalarImpl::Timestamptz(Timestamptz::from_micros(data.get_i64_le()))
         }
@@ -434,6 +443,13 @@ fn deserialize_timestamp(data: &mut impl Buf) -> Result<Timestamp> {
         .map_err(|_e| ValueEncodingError::InvalidTimestampEncoding(secs, nsecs))
 }
 
+fn deserialize_timestamp_nanosecond(data: &mut impl Buf) -> Result<TimestampNanosecond> {
+    let secs = data.get_i64_le();
+    let nsecs = data.get_u32_le();
+    TimestampNanosecond::with_secs_nsecs(secs, nsecs)
+        .map_err(|_e| ValueEncodingError::InvalidTimestampEncoding(secs, nsecs))
+}
+
 fn deserialize_date(data: &mut impl Buf) -> Result<Date> {
     let days = data.get_i32_le();
     Date::with_days_since_ce(days).map_err(|_e| ValueEncodingError::InvalidDateEncoding(days))
diff --git a/src/connector/src/parser/csv_parser.rs b/src/connector/src/parser/csv_parser.rs
index e81e4737803c6..72ad755acc4c9 100644
--- a/src/connector/src/parser/csv_parser.rs
+++ b/src/connector/src/parser/csv_parser.rs
@@ -13,7 +13,9 @@
 // limitations under the License.
 
 use risingwave_common::cast::str_to_bool;
-use risingwave_common::types::{Date, Decimal, ScalarImpl, Time, Timestamp, Timestamptz};
+use risingwave_common::types::{
+    Date, Decimal, ScalarImpl, Time, Timestamp, TimestampNanosecond, Timestamptz,
+};
 
 use super::unified::{AccessError, AccessResult};
 use super::{ByteStreamSourceParser, CsvProperties};
@@ -97,6 +99,7 @@ impl CsvParser {
             DataType::Date => parse!(v, Date)?.into(),
             DataType::Time => parse!(v, Time)?.into(),
             DataType::Timestamp => parse!(v, Timestamp)?.into(),
+            DataType::TimestampNanosecond => parse!(v, TimestampNanosecond)?.into(),
             DataType::Timestamptz => parse!(v, Timestamptz)?.into(),
             _ => {
                 return Err(AccessError::UnsupportedType {
diff --git a/src/connector/src/parser/mysql.rs b/src/connector/src/parser/mysql.rs
index 78a1685e8b733..a39b1c348c199 100644
--- a/src/connector/src/parser/mysql.rs
+++ b/src/connector/src/parser/mysql.rs
@@ -154,6 +154,15 @@ pub fn mysql_datum_to_rw_datum(
                 Timestamp
             )
         }
+        DataType::TimestampNanosecond => {
+            handle_data_type!(
+                mysql_row,
+                mysql_datum_index,
+                column_name,
+                chrono::NaiveDateTime,
+                Timestamp
+            )
+        }
         DataType::Timestamptz => {
             match mysql_row.take_opt::<Option<chrono::NaiveDateTime>, _>(mysql_datum_index) {
                 None => bail!(
diff --git a/src/connector/src/parser/postgres.rs b/src/connector/src/parser/postgres.rs
index 32e3c2c1e2e2c..598a0d973a4c2 100644
--- a/src/connector/src/parser/postgres.rs
+++ b/src/connector/src/parser/postgres.rs
@@ -68,6 +68,7 @@ fn postgres_cell_to_scalar_impl(
         | DataType::Date
         | DataType::Time
         | DataType::Timestamp
+        | DataType::TimestampNanosecond
         | DataType::Timestamptz
         | DataType::Jsonb
         | DataType::Interval
diff --git a/src/connector/src/parser/unified/json.rs b/src/connector/src/parser/unified/json.rs
index f19a56099d329..823f645dd2686 100644
--- a/src/connector/src/parser/unified/json.rs
+++ b/src/connector/src/parser/unified/json.rs
@@ -22,8 +22,8 @@ use risingwave_common::array::{ListValue, StructValue};
 use risingwave_common::cast::{i64_to_timestamp, i64_to_timestamptz, str_to_bytea};
 use risingwave_common::log::LogSuppresser;
 use risingwave_common::types::{
-    DataType, Date, Decimal, Int256, Interval, JsonbVal, ScalarImpl, Time, Timestamp, Timestamptz,
-    ToOwnedDatum,
+    DataType, Date, Decimal, Int256, Interval, JsonbVal, ScalarImpl, Time, Timestamp,
+    TimestampNanosecond, Timestamptz, ToOwnedDatum,
 };
 use risingwave_common::util::iter_util::ZipEqFast;
 use risingwave_connector_codec::decoder::utils::scaled_bigint_to_rust_decimal;
@@ -487,6 +487,12 @@ impl JsonParseOptions {
             ) => i64_to_timestamp(value.as_i64().unwrap())
                 .map_err(|_| create_error())?
                 .into(),
+            (DataType::TimestampNanosecond, ValueType::String) => value
+                .as_str()
+                .unwrap()
+                .parse::<TimestampNanosecond>()
+                .map_err(|_| create_error())?
+                .into(),
             // ---- Timestamptz -----
             (DataType::Timestamptz, ValueType::String) => match self.timestamptz_handling {
                 TimestamptzHandling::UtcWithoutSuffix => value
diff --git a/src/connector/src/sink/big_query.rs b/src/connector/src/sink/big_query.rs
index 18d2b5694e5db..3f0769180599a 100644
--- a/src/connector/src/sink/big_query.rs
+++ b/src/connector/src/sink/big_query.rs
@@ -325,6 +325,7 @@ impl BigQuerySink {
             DataType::Varchar => Ok("STRING".to_owned()),
             DataType::Time => Ok("TIME".to_owned()),
             DataType::Timestamp => Ok("DATETIME".to_owned()),
+            DataType::TimestampNanosecond => Ok("DATETIME".to_owned()),
             DataType::Timestamptz => Ok("TIMESTAMP".to_owned()),
             DataType::Interval => Ok("INTERVAL".to_owned()),
             DataType::Struct(structs) => {
@@ -369,6 +370,7 @@ impl BigQuerySink {
             DataType::Varchar => TableFieldSchema::string(&rw_field.name),
             DataType::Time => TableFieldSchema::time(&rw_field.name),
             DataType::Timestamp => TableFieldSchema::date_time(&rw_field.name),
+            DataType::TimestampNanosecond => TableFieldSchema::date_time(&rw_field.name),
             DataType::Timestamptz => TableFieldSchema::timestamp(&rw_field.name),
             DataType::Interval => {
                 return Err(SinkError::BigQuery(anyhow::anyhow!(
@@ -894,6 +896,9 @@ fn build_protobuf_field(
         DataType::Varchar => field.r#type = Some(field_descriptor_proto::Type::String.into()),
         DataType::Time => field.r#type = Some(field_descriptor_proto::Type::String.into()),
         DataType::Timestamp => field.r#type = Some(field_descriptor_proto::Type::String.into()),
+        DataType::TimestampNanosecond => {
+            field.r#type = Some(field_descriptor_proto::Type::String.into())
+        }
         DataType::Timestamptz => field.r#type = Some(field_descriptor_proto::Type::String.into()),
         DataType::Interval => field.r#type = Some(field_descriptor_proto::Type::String.into()),
         DataType::Struct(s) => {
diff --git a/src/connector/src/sink/clickhouse.rs b/src/connector/src/sink/clickhouse.rs
index 6a9c08330e58e..eca3f3ebaf49f 100644
--- a/src/connector/src/sink/clickhouse.rs
+++ b/src/connector/src/sink/clickhouse.rs
@@ -447,9 +447,12 @@ impl ClickHouseSink {
             risingwave_common::types::DataType::Time => Err(SinkError::ClickHouse(
                 "clickhouse can not support Time".to_owned(),
             )),
-            risingwave_common::types::DataType::Timestamp => Err(SinkError::ClickHouse(
-                "clickhouse does not have a type corresponding to naive timestamp".to_owned(),
-            )),
+            risingwave_common::types::DataType::Timestamp
+            | risingwave_common::types::DataType::TimestampNanosecond => {
+                Err(SinkError::ClickHouse(
+                    "clickhouse does not have a type corresponding to naive timestamp".to_owned(),
+                ))
+            }
             risingwave_common::types::DataType::Timestamptz => {
                 Ok(ck_column.r#type.contains("DateTime64"))
             }
@@ -945,7 +948,7 @@ impl ClickHouseFieldWithNull {
                     "clickhouse can not support Time".to_owned(),
                 ))
             }
-            ScalarRefImpl::Timestamp(_) => {
+            ScalarRefImpl::Timestamp(_) | ScalarRefImpl::TimestampNanosecond(_) => {
                 return Err(SinkError::ClickHouse(
                     "clickhouse does not have a type corresponding to naive timestamp".to_owned(),
                 ))
diff --git a/src/connector/src/sink/doris.rs b/src/connector/src/sink/doris.rs
index 6a0852f2fac08..b33f79f134b3e 100644
--- a/src/connector/src/sink/doris.rs
+++ b/src/connector/src/sink/doris.rs
@@ -170,7 +170,7 @@ impl DorisSink {
             risingwave_common::types::DataType::Time => {
                 Err(SinkError::Doris("doris can not support Time".to_owned()))
             }
-            risingwave_common::types::DataType::Timestamp => {
+            risingwave_common::types::DataType::Timestamp | risingwave_common::types::DataType::TimestampNanosecond => {
                 Ok(doris_data_type.contains("DATETIME"))
             }
             risingwave_common::types::DataType::Timestamptz => Err(SinkError::Doris(
diff --git a/src/connector/src/sink/dynamodb.rs b/src/connector/src/sink/dynamodb.rs
index bacee769b1a92..c18f84302d960 100644
--- a/src/connector/src/sink/dynamodb.rs
+++ b/src/connector/src/sink/dynamodb.rs
@@ -319,6 +319,7 @@ fn map_data(scalar_ref: Option<ScalarRefImpl<'_>>, data_type: &DataType) -> Resu
         | DataType::Time
         | DataType::Timestamp
         | DataType::Timestamptz
+        | DataType::TimestampNanosecond
         | DataType::Jsonb => AttributeValue::S(scalar_ref.to_text_with_type(data_type)),
         DataType::Boolean => AttributeValue::Bool(scalar_ref.into_bool()),
         DataType::Bytea => AttributeValue::B(Blob::new(scalar_ref.into_bytea())),
diff --git a/src/connector/src/sink/encoder/avro.rs b/src/connector/src/sink/encoder/avro.rs
index f61eab4d938d6..74070e993a620 100644
--- a/src/connector/src/sink/encoder/avro.rs
+++ b/src/connector/src/sink/encoder/avro.rs
@@ -461,7 +461,7 @@ fn on_field<D: MaybeData>(data_type: &DataType, maybe: D, expected: &AvroSchema)
             })?,
             _ => return no_match_err(),
         },
-        DataType::Timestamp => return no_match_err(),
+        DataType::Timestamp | DataType::TimestampNanosecond => return no_match_err(),
         DataType::Date => match inner {
             AvroSchema::Date => {
                 maybe.on_base(|s| Ok(Value::Date(s.into_date().get_nums_days_unix_epoch())))?
diff --git a/src/connector/src/sink/encoder/bson.rs b/src/connector/src/sink/encoder/bson.rs
index 7fb7ed571bff5..77630822a50c9 100644
--- a/src/connector/src/sink/encoder/bson.rs
+++ b/src/connector/src/sink/encoder/bson.rs
@@ -144,6 +144,9 @@ fn datum_to_bson(field: &Field, datum: DatumRef<'_>) -> Bson {
         (DataType::Timestamp, ScalarRefImpl::Timestamp(v)) => {
             Bson::DateTime(DateTime::from_millis(v.0.and_utc().timestamp_millis()))
         }
+        (DataType::TimestampNanosecond, ScalarRefImpl::TimestampNanosecond(v)) => {
+            Bson::DateTime(DateTime::from_millis(v.0.and_utc().timestamp_millis()))
+        }
         (DataType::Timestamptz, ScalarRefImpl::Timestamptz(v)) => {
             Bson::DateTime(DateTime::from_millis(v.timestamp_millis()))
         }
diff --git a/src/connector/src/sink/encoder/json.rs b/src/connector/src/sink/encoder/json.rs
index 248b1822b536b..a7f704da0b172 100644
--- a/src/connector/src/sink/encoder/json.rs
+++ b/src/connector/src/sink/encoder/json.rs
@@ -394,6 +394,7 @@ pub(crate) fn schema_type_mapping(rw_type: &DataType) -> &'static str {
         DataType::Varchar => "string",
         DataType::Time => "int64",
         DataType::Timestamp => "int64",
+        DataType::TimestampNanosecond => "int64",
         DataType::Timestamptz => "string",
         DataType::Interval => "string",
         DataType::Struct(_) => "struct",
diff --git a/src/connector/src/sink/encoder/proto.rs b/src/connector/src/sink/encoder/proto.rs
index 4d7ff35fa3bf2..516e42dbd36d7 100644
--- a/src/connector/src/sink/encoder/proto.rs
+++ b/src/connector/src/sink/encoder/proto.rs
@@ -423,6 +423,12 @@ fn on_field<D: MaybeData>(
             Kind::String => maybe.on_base(|s| Ok(Value::String(s.into_timestamp().to_string())))?,
             _ => return no_match_err(), // google.type.DateTime
         },
+        DataType::TimestampNanosecond => match proto_field.kind() {
+            Kind::String => {
+                maybe.on_base(|s| Ok(Value::String(s.into_timestampns().to_string())))?
+            }
+            _ => return no_match_err(), // google.type.DateTime
+        },
         DataType::Decimal => match proto_field.kind() {
             Kind::String => maybe.on_base(|s| Ok(Value::String(s.into_decimal().to_string())))?,
             _ => return no_match_err(), // google.type.Decimal
diff --git a/src/connector/src/sink/formatter/debezium_json.rs b/src/connector/src/sink/formatter/debezium_json.rs
index 3e076895c16b1..498af2491e080 100644
--- a/src/connector/src/sink/formatter/debezium_json.rs
+++ b/src/connector/src/sink/formatter/debezium_json.rs
@@ -298,7 +298,8 @@ pub(crate) fn field_to_json(field: &Field) -> Value {
         risingwave_common::types::DataType::Date => ("int32", "org.apache.kafka.connect.data.Date"),
         // may lose precision when Time/Timestamp has a fractional second precision value that is greater than 3. But for TimestampHandlingMode:Milli, it is sufficient.
         risingwave_common::types::DataType::Time => ("int64", "org.apache.kafka.connect.data.Time"),
-        risingwave_common::types::DataType::Timestamp => {
+        risingwave_common::types::DataType::Timestamp
+        | risingwave_common::types::DataType::TimestampNanosecond => {
             ("int64", "org.apache.kafka.connect.data.Timestamp")
         }
 
diff --git a/src/connector/src/sink/remote.rs b/src/connector/src/sink/remote.rs
index 582a5c861bb5c..0e4cb1187e3b1 100644
--- a/src/connector/src/sink/remote.rs
+++ b/src/connector/src/sink/remote.rs
@@ -187,6 +187,7 @@ async fn validate_remote_sink(param: &SinkParam, sink_name: &str) -> ConnectorRe
                     | DataType::Boolean
                     | DataType::Decimal
                     | DataType::Timestamp
+                    | DataType::TimestampNanosecond
                     | DataType::Timestamptz
                     | DataType::Varchar
                     | DataType::Date
diff --git a/src/connector/src/sink/sqlserver.rs b/src/connector/src/sink/sqlserver.rs
index 8314b2f45f0c5..a41d8944c34cd 100644
--- a/src/connector/src/sink/sqlserver.rs
+++ b/src/connector/src/sink/sqlserver.rs
@@ -580,6 +580,7 @@ fn bind_params(
                 },
                 ScalarRefImpl::Date(v) => query.bind(v.0),
                 ScalarRefImpl::Timestamp(v) => query.bind(v.0),
+                ScalarRefImpl::TimestampNanosecond(v) => query.bind(v.0),
                 ScalarRefImpl::Timestamptz(v) => query.bind(v.timestamp_micros()),
                 ScalarRefImpl::Time(v) => query.bind(v.0),
                 ScalarRefImpl::Bytea(v) => query.bind(v.to_vec()),
@@ -619,7 +620,7 @@ fn bind_params(
                 DataType::Time => {
                     query.bind(None as Option<chrono::NaiveTime>);
                 }
-                DataType::Timestamp => {
+                DataType::Timestamp | DataType::TimestampNanosecond => {
                     query.bind(None as Option<chrono::NaiveDateTime>);
                 }
                 DataType::Timestamptz => {
@@ -663,6 +664,7 @@ fn check_data_type_compatibility(data_type: &DataType) -> Result<()> {
         | DataType::Varchar
         | DataType::Time
         | DataType::Timestamp
+        | DataType::TimestampNanosecond
         | DataType::Timestamptz
         | DataType::Bytea => Ok(()),
         DataType::Interval => Err(data_type_not_supported("Interval")),
diff --git a/src/connector/src/sink/starrocks.rs b/src/connector/src/sink/starrocks.rs
index 65c6348dbaf44..831cb4ae59103 100644
--- a/src/connector/src/sink/starrocks.rs
+++ b/src/connector/src/sink/starrocks.rs
@@ -218,7 +218,7 @@ impl StarrocksSink {
             risingwave_common::types::DataType::Time => Err(SinkError::Starrocks(
                 "TIME is not supported for Starrocks sink. Please convert to VARCHAR or other supported types.".to_owned(),
             )),
-            risingwave_common::types::DataType::Timestamp => {
+            risingwave_common::types::DataType::Timestamp | risingwave_common::types::DataType::TimestampNanosecond => {
                 Ok(starrocks_data_type.contains("datetime"))
             }
             risingwave_common::types::DataType::Timestamptz => Err(SinkError::Starrocks(
diff --git a/src/expr/impl/src/aggregate/general.rs b/src/expr/impl/src/aggregate/general.rs
index 83a645122301e..b6f36e69085eb 100644
--- a/src/expr/impl/src/aggregate/general.rs
+++ b/src/expr/impl/src/aggregate/general.rs
@@ -115,6 +115,7 @@ fn min<T: Ord>(state: T, input: T) -> T {
 #[aggregate("max(time) -> auto", state = "ref")]
 #[aggregate("max(interval) -> auto", state = "ref")]
 #[aggregate("max(timestamp) -> auto", state = "ref")]
+#[aggregate("max(timestamp_ns) -> auto", state = "ref")]
 #[aggregate("max(timestamptz) -> auto", state = "ref")]
 #[aggregate("max(varchar) -> auto", state = "ref")]
 #[aggregate("max(bytea) -> auto", state = "ref")]
diff --git a/src/expr/impl/src/scalar/cast.rs b/src/expr/impl/src/scalar/cast.rs
index 66d141ca05172..db76d7671d574 100644
--- a/src/expr/impl/src/scalar/cast.rs
+++ b/src/expr/impl/src/scalar/cast.rs
@@ -37,6 +37,7 @@ use thiserror_ext::AsReport;
 #[function("cast(varchar) -> timestamp")]
 #[function("cast(varchar) -> interval")]
 #[function("cast(varchar) -> jsonb")]
+#[function("cast(varchar) -> timestamp_ns")]
 pub fn str_parse<T>(elem: &str, ctx: &Context) -> Result<T>
 where
     T: FromStr,
@@ -125,9 +126,14 @@ where
 #[function("cast(int8) -> decimal")]
 #[function("cast(float4) -> float8")]
 #[function("cast(date) -> timestamp")]
+#[function("cast(date) -> timestamp_ns")]
 #[function("cast(time) -> interval")]
 #[function("cast(timestamp) -> date")]
+#[function("cast(timestamp_ns) -> date")]
 #[function("cast(timestamp) -> time")]
+#[function("cast(timestamp_ns) -> time")]
+#[function("cast(timestamp_ns) -> timestamp")]
+#[function("cast(timestamp) -> timestamp_ns")]
 #[function("cast(interval) -> time")]
 #[function("cast(varchar) -> varchar")]
 #[function("cast(int256) -> float8")]
@@ -158,6 +164,7 @@ pub fn int_to_bool(input: i32) -> bool {
 #[function("cast(date) -> varchar")]
 #[function("cast(interval) -> varchar")]
 #[function("cast(timestamp) -> varchar")]
+#[function("cast(timestamp_ns) -> varchar")]
 #[function("cast(jsonb) -> varchar")]
 #[function("cast(bytea) -> varchar")]
 #[function("cast(anyarray) -> varchar")]
diff --git a/src/expr/impl/src/scalar/cmp.rs b/src/expr/impl/src/scalar/cmp.rs
index d8ae863b65124..4fe852550c15e 100644
--- a/src/expr/impl/src/scalar/cmp.rs
+++ b/src/expr/impl/src/scalar/cmp.rs
@@ -92,6 +92,7 @@ where
 #[function("greater_than_or_equal(time, time) -> boolean")]
 #[function("greater_than_or_equal(interval, interval) -> boolean")]
 #[function("greater_than_or_equal(timestamp, timestamp) -> boolean")]
+#[function("greater_than_or_equal(timestamp_ns, timestamp_ns) -> boolean")]
 #[function("greater_than_or_equal(timestamptz, timestamptz) -> boolean")]
 #[function("greater_than_or_equal(date, timestamp) -> boolean")]
 #[function("greater_than_or_equal(timestamp, date) -> boolean")]
diff --git a/src/expr/impl/src/scalar/extract.rs b/src/expr/impl/src/scalar/extract.rs
index c2f467ce482fb..c6c3894f22338 100644
--- a/src/expr/impl/src/scalar/extract.rs
+++ b/src/expr/impl/src/scalar/extract.rs
@@ -15,7 +15,9 @@
 use std::str::FromStr;
 
 use chrono::{Datelike, NaiveTime, Timelike};
-use risingwave_common::types::{Date, Decimal, Interval, Time, Timestamp, Timestamptz, F64};
+use risingwave_common::types::{
+    Date, Decimal, Interval, Time, Timestamp, TimestampNanosecond, Timestamptz, F64,
+};
 use risingwave_expr::{function, ExprError, Result};
 
 use self::Unit::*;
@@ -93,24 +95,45 @@ fn extract_from_time(time: Time, unit: &Unit) -> Decimal {
 fn extract_from_timestamp(timestamp: Timestamp, unit: &Unit) -> Decimal {
     match unit {
         Epoch => {
-            if let Some(nanos) = timestamp.0.and_utc().timestamp_nanos_opt() {
+            let micros = timestamp.0.and_utc().timestamp_micros();
+            Decimal::from_i128_with_scale(micros as i128, 6)
+        }
+        Julian => {
+            let epoch =
+                Decimal::from_i128_with_scale(timestamp.0.and_utc().timestamp_micros() as i128, 6);
+            epoch / (24 * 60 * 60).into() + 2_440_588.into()
+        }
+        _ if unit.is_date_unit() => extract_from_datelike(timestamp.0.date(), *unit),
+        _ if unit.is_time_unit() => extract_from_timelike(timestamp.0.time(), *unit),
+        u => unreachable!("invalid unit {:?} for timestamp", u),
+    }
+}
+
+#[function(
+    "extract(varchar, timestamp_ns) -> decimal",
+    prebuild = "Unit::from_str($0)?.ensure_timestamp()?"
+)]
+fn extract_from_timestamp_ns(timestamp_nano: TimestampNanosecond, unit: &Unit) -> Decimal {
+    match unit {
+        Epoch => {
+            if let Some(nanos) = timestamp_nano.0.and_utc().timestamp_nanos_opt() {
                 Decimal::from_i128_with_scale(nanos as i128, 9)
             } else {
-                let micros = timestamp.0.and_utc().timestamp_micros();
+                let micros = timestamp_nano.0.and_utc().timestamp_micros();
                 Decimal::from_i128_with_scale(micros as i128, 6)
             }
         }
         Julian => {
-            let epoch = if let Some(nanos) = timestamp.0.and_utc().timestamp_nanos_opt() {
+            let epoch = if let Some(nanos) = timestamp_nano.0.and_utc().timestamp_nanos_opt() {
                 Decimal::from_i128_with_scale(nanos as i128, 9)
             } else {
-                let epoch = timestamp.0.and_utc().timestamp_micros();
+                let epoch = timestamp_nano.0.and_utc().timestamp_micros();
                 Decimal::from_i128_with_scale(epoch as i128, 6)
             };
             epoch / (24 * 60 * 60).into() + 2_440_588.into()
         }
-        _ if unit.is_date_unit() => extract_from_datelike(timestamp.0.date(), *unit),
-        _ if unit.is_time_unit() => extract_from_timelike(timestamp.0.time(), *unit),
+        _ if unit.is_date_unit() => extract_from_datelike(timestamp_nano.0.date(), *unit),
+        _ if unit.is_time_unit() => extract_from_timelike(timestamp_nano.0.time(), *unit),
         u => unreachable!("invalid unit {:?} for timestamp", u),
     }
 }
diff --git a/src/expr/impl/src/scalar/make_time.rs b/src/expr/impl/src/scalar/make_time.rs
index 9d1edd1adc149..3a0e959a3a110 100644
--- a/src/expr/impl/src/scalar/make_time.rs
+++ b/src/expr/impl/src/scalar/make_time.rs
@@ -13,7 +13,9 @@
 // limitations under the License.
 
 use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
-use risingwave_common::types::{Date, FloatExt, Time, Timestamp, Timestamptz, F64};
+use risingwave_common::types::{
+    Date, FloatExt, Time, Timestamp, TimestampNanosecond, Timestamptz, F64,
+};
 use risingwave_expr::expr_context::TIME_ZONE;
 use risingwave_expr::{capture_context, function, ExprError, Result};
 
@@ -35,7 +37,7 @@ pub fn make_naive_date(mut year: i32, month: i32, day: i32) -> Result<NaiveDate>
     })
 }
 
-fn make_naive_time(hour: i32, min: i32, sec: F64) -> Result<NaiveTime> {
+fn make_naive_time_ns(hour: i32, min: i32, sec: F64) -> Result<NaiveTime> {
     if !sec.is_finite() || sec.0.is_sign_negative() {
         return Err(ExprError::InvalidParam {
             name: "sec",
@@ -52,6 +54,23 @@ fn make_naive_time(hour: i32, min: i32, sec: F64) -> Result<NaiveTime> {
     )
 }
 
+fn make_naive_time_ms(hour: i32, min: i32, sec: F64) -> Result<NaiveTime> {
+    if !sec.is_finite() || sec.0.is_sign_negative() {
+        return Err(ExprError::InvalidParam {
+            name: "sec",
+            reason: format!("invalid sec: {}", sec).into(),
+        });
+    }
+    let sec_u32 = sec.0.trunc() as u32;
+    let microsecond_u32 = ((sec.0 - sec.0.trunc()) * 1_000_000.0).round_ties_even() as u32;
+    NaiveTime::from_hms_micro_opt(hour as u32, min as u32, sec_u32, microsecond_u32).ok_or_else(
+        || ExprError::InvalidParam {
+            name: "hour, min, sec",
+            reason: format!("invalid time: {}:{}:{}", hour, min, sec).into(),
+        },
+    )
+}
+
 // year int, month int, day int
 #[function("make_date(int4, int4, int4) -> date")]
 pub fn make_date(year: i32, month: i32, day: i32) -> Result<Date> {
@@ -61,7 +80,7 @@ pub fn make_date(year: i32, month: i32, day: i32) -> Result<Date> {
 // hour int, min int, sec double precision
 #[function("make_time(int4, int4, float8) -> time")]
 pub fn make_time(hour: i32, min: i32, sec: F64) -> Result<Time> {
-    Ok(Time(make_naive_time(hour, min, sec)?))
+    Ok(Time(make_naive_time_ns(hour, min, sec)?))
 }
 
 // year int, month int, day int, hour int, min int, sec double precision
@@ -76,7 +95,22 @@ pub fn make_timestamp(
 ) -> Result<Timestamp> {
     Ok(Timestamp(NaiveDateTime::new(
         make_naive_date(year, month, day)?,
-        make_naive_time(hour, min, sec)?,
+        make_naive_time_ms(hour, min, sec)?,
+    )))
+}
+
+#[function("make_timestamp_ns(int4, int4, int4, int4, int4, float8) -> timestamp_ns")]
+pub fn make_timestamp_ns(
+    year: i32,
+    month: i32,
+    day: i32,
+    hour: i32,
+    min: i32,
+    sec: F64,
+) -> Result<TimestampNanosecond> {
+    Ok(TimestampNanosecond(NaiveDateTime::new(
+        make_naive_date(year, month, day)?,
+        make_naive_time_ns(hour, min, sec)?,
     )))
 }
 
@@ -119,7 +153,7 @@ fn make_timestamptz_impl(
 ) -> Result<Timestamptz> {
     let naive_date_time = NaiveDateTime::new(
         make_naive_date(year, month, day)?,
-        make_naive_time(hour, min, sec)?,
+        make_naive_time_ms(hour, min, sec)?,
     );
     timestamp_at_time_zone(Timestamp(naive_date_time), time_zone)
 }
diff --git a/src/expr/impl/src/scalar/to_char.rs b/src/expr/impl/src/scalar/to_char.rs
index 9bfc73be0bf7f..5ad078a69ecda 100644
--- a/src/expr/impl/src/scalar/to_char.rs
+++ b/src/expr/impl/src/scalar/to_char.rs
@@ -18,7 +18,7 @@ use std::sync::LazyLock;
 use aho_corasick::{AhoCorasick, AhoCorasickBuilder};
 use chrono::format::{Item, StrftimeItems};
 use chrono::{Datelike, NaiveDate};
-use risingwave_common::types::{Interval, Timestamp, Timestamptz};
+use risingwave_common::types::{Interval, Timestamp, TimestampNanosecond, Timestamptz};
 use risingwave_expr::{function, ExprError, Result};
 
 use super::timestamptz::time_zone_err;
@@ -194,6 +194,19 @@ fn timestamp_to_char(data: Timestamp, pattern: &ChronoPattern, writer: &mut impl
     write!(writer, "{}", format).unwrap();
 }
 
+#[function(
+    "to_char(timestamp_ns, varchar) -> varchar",
+    prebuild = "ChronoPattern::compile($1)"
+)]
+fn timestamp_ns_to_char(
+    data: TimestampNanosecond,
+    pattern: &ChronoPattern,
+    writer: &mut impl Write,
+) {
+    let format = data.0.format_with_items(pattern.borrow_dependent().iter());
+    write!(writer, "{}", format).unwrap();
+}
+
 #[function("to_char(timestamptz, varchar) -> varchar", rewritten)]
 fn _timestamptz_to_char() {}
 
diff --git a/src/expr/impl/src/scalar/to_jsonb.rs b/src/expr/impl/src/scalar/to_jsonb.rs
index 18b89e946f75c..c934b3407a905 100644
--- a/src/expr/impl/src/scalar/to_jsonb.rs
+++ b/src/expr/impl/src/scalar/to_jsonb.rs
@@ -17,7 +17,8 @@ use std::fmt::Debug;
 use jsonbb::Builder;
 use risingwave_common::types::{
     DataType, Date, Decimal, Int256Ref, Interval, JsonbRef, JsonbVal, ListRef, MapRef,
-    ScalarRefImpl, Serial, StructRef, Time, Timestamp, Timestamptz, ToText, F32, F64,
+    ScalarRefImpl, Serial, StructRef, Time, Timestamp, TimestampNanosecond, Timestamptz, ToText,
+    F32, F64,
 };
 use risingwave_common::util::iter_util::ZipEqDebug;
 use risingwave_expr::expr::Context;
@@ -66,6 +67,7 @@ impl ToJsonb for ScalarRefImpl<'_> {
             Date(v) => v.add_to(ty, builder),
             Time(v) => v.add_to(ty, builder),
             Timestamp(v) => v.add_to(ty, builder),
+            TimestampNanosecond(v) => v.add_to(ty, builder),
             Jsonb(v) => v.add_to(ty, builder),
             Serial(v) => v.add_to(ty, builder),
             Bytea(v) => v.add_to(ty, builder),
@@ -194,6 +196,13 @@ impl ToJsonb for Timestamp {
     }
 }
 
+impl ToJsonb for TimestampNanosecond {
+    fn add_to(self, _: &DataType, builder: &mut Builder) -> Result<()> {
+        builder.display(format_args!("{}T{}", self.0.date(), self.0.time()));
+        Ok(())
+    }
+}
+
 impl ToJsonb for Timestamptz {
     fn add_to(self, _: &DataType, builder: &mut Builder) -> Result<()> {
         let instant_local = self.to_datetime_utc();
diff --git a/src/expr/impl/src/table_function/generate_series.rs b/src/expr/impl/src/table_function/generate_series.rs
index 3558a3cf7f20b..61869206681d5 100644
--- a/src/expr/impl/src/table_function/generate_series.rs
+++ b/src/expr/impl/src/table_function/generate_series.rs
@@ -36,6 +36,7 @@ where
 #[function("generate_series(int4, int4, int4) -> setof int4")]
 #[function("generate_series(int8, int8, int8) -> setof int8")]
 #[function("generate_series(timestamp, timestamp, interval) -> setof timestamp")]
+#[function("generate_series(timestamp_ns, timestamp_ns, interval) -> setof timestamp_ns")]
 fn generate_series_step<T, S>(start: T, stop: T, step: S) -> Result<impl Iterator<Item = T>>
 where
     T: CheckedAdd<S, Output = T> + PartialOrd + Copy,
diff --git a/src/expr/impl/src/udf/wasm.rs b/src/expr/impl/src/udf/wasm.rs
index a3ee372d4f95f..0d11491bb6df4 100644
--- a/src/expr/impl/src/udf/wasm.rs
+++ b/src/expr/impl/src/udf/wasm.rs
@@ -264,6 +264,7 @@ fn datatype_name(ty: &DataType) -> String {
         DataType::Date => "date32".to_owned(),
         DataType::Time => "time64".to_owned(),
         DataType::Timestamp => "timestamp".to_owned(),
+        DataType::TimestampNanosecond => "timestamp_ns".to_owned(),
         DataType::Timestamptz => "timestamptz".to_owned(),
         DataType::Interval => "interval".to_owned(),
         DataType::Decimal => "decimal".to_owned(),
diff --git a/src/expr/impl/tests/sig.rs b/src/expr/impl/tests/sig.rs
index 59a76c467157b..0e3ea5ffa107c 100644
--- a/src/expr/impl/tests/sig.rs
+++ b/src/expr/impl/tests/sig.rs
@@ -62,8 +62,8 @@ fn test_func_sig_map() {
             "cast(anyarray) -> character varying/anyarray",
             "cast(bigint) -> rw_int256/serial/integer/smallint/numeric/double precision/real/character varying",
             "cast(boolean) -> integer/character varying",
-            "cast(character varying) -> jsonb/interval/timestamp without time zone/time without time zone/date/rw_int256/real/double precision/numeric/smallint/integer/bigint/character varying/boolean/bytea/anyarray",
-            "cast(date) -> timestamp without time zone/character varying",
+            "cast(character varying) -> timestamp_ns without time zone/jsonb/interval/timestamp without time zone/time without time zone/date/rw_int256/real/double precision/numeric/smallint/integer/bigint/character varying/boolean/bytea/anyarray",
+            "cast(date) -> timestamp_ns without time zone/timestamp without time zone/character varying",
             "cast(double precision) -> numeric/real/bigint/integer/smallint/character varying",
             "cast(integer) -> rw_int256/smallint/numeric/double precision/real/bigint/boolean/character varying",
             "cast(interval) -> time without time zone/character varying",
@@ -74,6 +74,7 @@ fn test_func_sig_map() {
             "cast(smallint) -> rw_int256/numeric/double precision/real/bigint/integer/character varying",
             "cast(time without time zone) -> interval/character varying",
             "cast(timestamp without time zone) -> time without time zone/date/character varying",
+            "cast(timestamp_ns without time zone) -> time without time zone/date/character varying",
             "greatest() -> bytea/character varying/timestamp with time zone/timestamp without time zone/interval/time without time zone/date/rw_int256/serial/real/double precision/numeric/smallint/integer/bigint/boolean",
             "least() -> bytea/character varying/timestamp with time zone/timestamp without time zone/interval/time without time zone/date/rw_int256/serial/real/double precision/numeric/smallint/integer/bigint/boolean",
         ]
diff --git a/src/expr/macro/src/lib.rs b/src/expr/macro/src/lib.rs
index 6e643d097bf65..2eee3f75805de 100644
--- a/src/expr/macro/src/lib.rs
+++ b/src/expr/macro/src/lib.rs
@@ -386,6 +386,7 @@ mod utils;
 /// | date        | `date`             | `Date`        | `Date`             | yes        |
 /// | time        | `time`             | `Time`        | `Time`             | yes        |
 /// | timestamp   | `timestamp`        | `Timestamp`   | `Timestamp`        | yes        |
+/// | timestamp_ns   | `timestamp_ns`        | `TimestampNanosecond`   | `TimestampNanosecond`        | yes        |
 /// | timestamptz | `timestamptz`      | `Timestamptz` | `Timestamptz`      | yes        |
 /// | interval    | `interval`         | `Interval`    | `Interval`         | yes        |
 /// | varchar     | `varchar`          | `Box<str>`    | `&str`             | no         |
diff --git a/src/expr/macro/src/types.rs b/src/expr/macro/src/types.rs
index a696c4d72de33..0c8ac019c7c0a 100644
--- a/src/expr/macro/src/types.rs
+++ b/src/expr/macro/src/types.rs
@@ -28,6 +28,7 @@ const TYPE_MATRIX: &str = "
     date        Date        DateArray           Date            Date                y
     time        Time        TimeArray           Time            Time                y
     timestamp   Timestamp   TimestampArray      Timestamp       Timestamp           y
+    timestamp_ns   TimestampNanosecond   TimestampNanosecondArray      TimestampNanosecond       TimestampNanosecond           y
     timestamptz Timestamptz TimestamptzArray    Timestamptz     Timestamptz         y
     interval    Interval    IntervalArray       Interval        Interval            y
     varchar     Varchar     Utf8Array           Box<str>        &str                _
diff --git a/src/frontend/src/binder/expr/function/builtin_scalar.rs b/src/frontend/src/binder/expr/function/builtin_scalar.rs
index 7b75689895c41..85cc1efda75d0 100644
--- a/src/frontend/src/binder/expr/function/builtin_scalar.rs
+++ b/src/frontend/src/binder/expr/function/builtin_scalar.rs
@@ -223,6 +223,7 @@ impl Binder {
                 ("make_date", raw_call(ExprType::MakeDate)),
                 ("make_time", raw_call(ExprType::MakeTime)),
                 ("make_timestamp", raw_call(ExprType::MakeTimestamp)),
+                ("make_timestampns", raw_call(ExprType::MakeTimestampNs)),
                 ("make_timestamptz", raw_call(ExprType::MakeTimestamptz)),
                 ("timezone", rewrite(ExprType::AtTimeZone, |mut inputs|{
                     if inputs.len() == 2 {
diff --git a/src/frontend/src/binder/expr/mod.rs b/src/frontend/src/binder/expr/mod.rs
index 2002555c857be..39f996469f24c 100644
--- a/src/frontend/src/binder/expr/mod.rs
+++ b/src/frontend/src/binder/expr/mod.rs
@@ -1062,6 +1062,7 @@ pub fn bind_data_type(data_type: &AstDataType) -> Result<DataType> {
                 "float4" => DataType::Float32,
                 "float8" => DataType::Float64,
                 "timestamptz" => DataType::Timestamptz,
+                "timestamp_ns" => DataType::TimestampNanosecond,
                 "text" => DataType::Varchar,
                 "serial" => {
                     return Err(ErrorCode::NotSupported(
diff --git a/src/frontend/src/expr/literal.rs b/src/frontend/src/expr/literal.rs
index 34b742abe5b45..e5a6701a96c5f 100644
--- a/src/frontend/src/expr/literal.rs
+++ b/src/frontend/src/expr/literal.rs
@@ -49,6 +49,7 @@ impl std::fmt::Debug for Literal {
                     | DataType::Bytea
                     | DataType::Date
                     | DataType::Timestamp
+                    | DataType::TimestampNanosecond
                     | DataType::Timestamptz
                     | DataType::Time
                     | DataType::Interval
diff --git a/src/frontend/src/expr/pure.rs b/src/frontend/src/expr/pure.rs
index 013f9f1d7ddb1..d911d46237d6c 100644
--- a/src/frontend/src/expr/pure.rs
+++ b/src/frontend/src/expr/pure.rs
@@ -67,6 +67,7 @@ impl ExprVisitor for ImpureAnalyzer {
             | Type::MakeDate
             | Type::MakeTime
             | Type::MakeTimestamp
+            | Type::MakeTimestampNs
             | Type::CharToTimestamptz
             | Type::CharToDate
             | Type::CastWithTimeZone
diff --git a/src/frontend/src/expr/session_timezone.rs b/src/frontend/src/expr/session_timezone.rs
index 11423c4dc138d..e37f9580de634 100644
--- a/src/frontend/src/expr/session_timezone.rs
+++ b/src/frontend/src/expr/session_timezone.rs
@@ -104,13 +104,15 @@ impl SessionTimezone {
                         Some(self.cast_with_timezone(input, return_type))
                     }
                     (DataType::Date, DataType::Timestamptz)
-                    | (DataType::Timestamp, DataType::Timestamptz) => {
+                    | (DataType::Timestamp, DataType::Timestamptz)
+                    | (DataType::TimestampNanosecond, DataType::Timestamptz) => {
                         input = input.cast_explicit(DataType::Timestamp).unwrap();
                         Some(self.at_timezone(input))
                     }
                     (DataType::Timestamptz, DataType::Date)
                     | (DataType::Timestamptz, DataType::Time)
-                    | (DataType::Timestamptz, DataType::Timestamp) => {
+                    | (DataType::Timestamptz, DataType::Timestamp)
+                    | (DataType::Timestamptz, DataType::TimestampNanosecond) => {
                         input = self.at_timezone(input);
                         input = input.cast_explicit(return_type).unwrap();
                         Some(input)
@@ -140,7 +142,7 @@ impl SessionTimezone {
                     if matches!(inputs[(idx + 1) % 2].return_type(), DataType::Timestamptz)
                         && matches!(
                             inputs[idx % 2].return_type(),
-                            DataType::Date | DataType::Timestamp
+                            DataType::Date | DataType::Timestamp | DataType::TimestampNanosecond
                         )
                     {
                         let mut to_cast = inputs[idx % 2].clone();
diff --git a/src/frontend/src/expr/type_inference/cast.rs b/src/frontend/src/expr/type_inference/cast.rs
index 949c24cea1c90..9a5fb5ecdc897 100644
--- a/src/frontend/src/expr/type_inference/cast.rs
+++ b/src/frontend/src/expr/type_inference/cast.rs
@@ -306,23 +306,24 @@ pub static CAST_TABLE: LazyLock<CastTable> = LazyLock::new(|| {
     use DataTypeName::*;
     const CAST_TABLE: &[(&str, DataTypeName)] = &[
         // 123456789ABCDEF
-        (". e            a ", Boolean),     // 0
-        (" .iiiiii       a ", Int16),       // 1
-        ("ea.iiiii       a ", Int32),       // 2
-        (" aa.iiii       ae", Int64),       // 3
-        (" aaa.ii        a ", Decimal),     // 4
-        (" aaaa.i        a ", Float32),     // 5
-        (" aaaaa.        a ", Float64),     // 6
-        ("      e.       a ", Int256),      // 7
-        ("        .ii    a ", Date),        // 8
-        ("        a.ia   a ", Timestamp),   // 9
-        ("        aa.a   a ", Timestamptz), // A
-        ("           .i  a ", Time),        // B
-        ("           a.  a ", Interval),    // C
-        ("eeeeeee      . a ", Jsonb),       // D
-        ("              .a ", Bytea),       // E
-        ("eeeeeeeeeeeeeee. ", Varchar),     // F
-        ("   e            .", Serial),
+        (". e            a  ", Boolean),             // 0
+        (" .iiiiii       a  ", Int16),               // 1
+        ("ea.iiiii       a  ", Int32),               // 2
+        (" aa.iiii       ae ", Int64),               // 3
+        (" aaa.ii        a  ", Decimal),             // 4
+        (" aaaa.i        a  ", Float32),             // 5
+        (" aaaaa.        a  ", Float64),             // 6
+        ("      e.       a  ", Int256),              // 7
+        ("        .ii    a i", Date),                // 8
+        ("        a.ia   a i", Timestamp),           // 9
+        ("        aa.a   a a", Timestamptz),         // A
+        ("           .i  a  ", Time),                // B
+        ("           a.  a  ", Interval),            // C
+        ("eeeeeee      . a  ", Jsonb),               // D
+        ("              .a  ", Bytea),               // E
+        ("eeeeeeeeeeeeeee. e", Varchar),             // F
+        ("   e            . ", Serial),              // G
+        ("        aaia   a .", TimestampNanosecond), // H
     ];
     let mut map = BTreeMap::new();
     for (row, source) in CAST_TABLE {
diff --git a/src/frontend/src/optimizer/plan_expr_visitor/strong.rs b/src/frontend/src/optimizer/plan_expr_visitor/strong.rs
index 6a0ac319c1db0..dce444820bc23 100644
--- a/src/frontend/src/optimizer/plan_expr_visitor/strong.rs
+++ b/src/frontend/src/optimizer/plan_expr_visitor/strong.rs
@@ -126,6 +126,7 @@ impl Strong {
             | ExprType::TumbleStart
             | ExprType::MakeDate
             | ExprType::MakeTime
+            | ExprType::MakeTimestampNs
             | ExprType::MakeTimestamp
             | ExprType::SecToTimestamptz
             | ExprType::AtTimeZone
diff --git a/src/frontend/src/optimizer/property/monotonicity.rs b/src/frontend/src/optimizer/property/monotonicity.rs
index db020cdb2c64c..e610e5472df3a 100644
--- a/src/frontend/src/optimizer/property/monotonicity.rs
+++ b/src/frontend/src/optimizer/property/monotonicity.rs
@@ -232,7 +232,8 @@ impl MonotonicityAnalyzer {
                     // the result should have the same monotonicity.
                     // 2. For at_time_zone(timestamptz, const timezone) -> timestamp, when timestamptz has some monotonicity,
                     // the result only have the same monotonicity when the timezone is without DST (Daylight Saving Time).
-                    if (func_call.inputs()[0].return_type() == DataType::Timestamp
+                    if ((func_call.inputs()[0].return_type() == DataType::Timestamp
+                        || func_call.inputs()[0].return_type() == DataType::TimestampNanosecond)
                         && func_call.return_type() == DataType::Timestamptz)
                         || time_zone_is_without_dst(time_zone)
                     {
diff --git a/src/frontend/src/optimizer/rule/index_selection_rule.rs b/src/frontend/src/optimizer/rule/index_selection_rule.rs
index 3ff8efcd281c0..48fecd4d11bc8 100644
--- a/src/frontend/src/optimizer/rule/index_selection_rule.rs
+++ b/src/frontend/src/optimizer/rule/index_selection_rule.rs
@@ -54,7 +54,8 @@ use std::rc::Rc;
 use itertools::Itertools;
 use risingwave_common::catalog::Schema;
 use risingwave_common::types::{
-    DataType, Date, Decimal, Int256, Interval, Serial, Time, Timestamp, Timestamptz,
+    DataType, Date, Decimal, Int256, Interval, Serial, Time, Timestamp, TimestampNanosecond,
+    Timestamptz,
 };
 use risingwave_common::util::iter_util::ZipEqFast;
 use risingwave_pb::plan_common::JoinType;
@@ -763,6 +764,7 @@ impl<'a> TableScanIoEstimator<'a> {
             DataType::Date => size_of::<Date>(),
             DataType::Time => size_of::<Time>(),
             DataType::Timestamp => size_of::<Timestamp>(),
+            DataType::TimestampNanosecond => size_of::<TimestampNanosecond>(),
             DataType::Timestamptz => size_of::<Timestamptz>(),
             DataType::Interval => size_of::<Interval>(),
             DataType::Int256 => Int256::size(),
diff --git a/src/frontend/src/utils/data_type.rs b/src/frontend/src/utils/data_type.rs
index e9cb98b77f164..e27bd74b67396 100644
--- a/src/frontend/src/utils/data_type.rs
+++ b/src/frontend/src/utils/data_type.rs
@@ -32,6 +32,9 @@ impl DataType {
             DataType::Varchar => AstDataType::Varchar,
             DataType::Time => AstDataType::Time(false),
             DataType::Timestamp => AstDataType::Timestamp(false),
+            DataType::TimestampNanosecond => {
+                AstDataType::Custom(vec!["timestamp_ns".into()].into())
+            }
             DataType::Timestamptz => AstDataType::Timestamp(true),
             DataType::Interval => AstDataType::Interval,
             DataType::Jsonb => AstDataType::Jsonb,
diff --git a/src/tests/sqlsmith/src/sql_gen/types.rs b/src/tests/sqlsmith/src/sql_gen/types.rs
index 92e13810f5e21..882823d269fb2 100644
--- a/src/tests/sqlsmith/src/sql_gen/types.rs
+++ b/src/tests/sqlsmith/src/sql_gen/types.rs
@@ -39,6 +39,7 @@ pub(super) fn data_type_to_ast_data_type(data_type: &DataType) -> AstDataType {
         DataType::Bytea => AstDataType::Bytea,
         DataType::Date => AstDataType::Date,
         DataType::Timestamp => AstDataType::Timestamp(false),
+        DataType::TimestampNanosecond => AstDataType::Timestamp(false),
         DataType::Timestamptz => AstDataType::Timestamp(true),
         DataType::Time => AstDataType::Time(false),
         DataType::Interval => AstDataType::Interval,