From 8a89615ddb63070d6ce3420c54f281c41677d76c Mon Sep 17 00:00:00 2001
From: xxhZs <1060434431@qq.com>
Date: Wed, 19 Feb 2025 16:49:59 +0800
Subject: [PATCH] fix comm

---
 proto/data.proto                              |   4 +-
 .../executors/src/executor/postgres_query.rs  |   1 -
 src/common/src/array/arrow/arrow_iceberg.rs   |   4 +-
 src/common/src/array/arrow/arrow_impl.rs      |  42 +++---
 src/common/src/array/chrono_array.rs          |   6 +-
 src/common/src/array/mod.rs                   |   4 +-
 src/common/src/array/primitive_array.rs       |   2 +-
 src/common/src/array/proto_reader.rs          |   4 +-
 src/common/src/hash/key.rs                    |   8 +-
 src/common/src/row/owned_row.rs               |   8 +-
 src/common/src/test_utils/rand_array.rs       |   6 +-
 src/common/src/test_utils/rand_chunk.rs       |   8 +-
 src/common/src/types/datetime.rs              | 134 ++++++++++--------
 src/common/src/types/macros.rs                |   2 +-
 src/common/src/types/map_type.rs              |   2 +-
 src/common/src/types/mod.rs                   |  39 +++--
 src/common/src/types/postgres_type.rs         |   8 +-
 src/common/src/types/scalar_impl.rs           |  12 +-
 src/common/src/types/to_binary.rs             |   6 +-
 src/common/src/types/to_sql.rs                |   4 +-
 src/common/src/types/with_data_type.rs        |   4 +-
 src/common/src/util/memcmp_encoding.rs        |   6 +-
 src/common/src/util/value_encoding/mod.rs     |  12 +-
 src/connector/src/parser/csv_parser.rs        |   4 +-
 src/connector/src/parser/mysql.rs             |   6 +-
 src/connector/src/parser/postgres.rs          |   3 +-
 src/connector/src/parser/unified/json.rs      |   8 +-
 src/connector/src/sink/big_query.rs           |  28 ++--
 src/connector/src/sink/clickhouse.rs          |  10 +-
 src/connector/src/sink/doris.rs               |  15 +-
 src/connector/src/sink/dynamodb.rs            |   2 +-
 src/connector/src/sink/encoder/avro.rs        |   2 +-
 src/connector/src/sink/encoder/bson.rs        |   2 +-
 src/connector/src/sink/encoder/json.rs        |   2 +-
 src/connector/src/sink/encoder/proto.rs       |   4 +-
 .../src/sink/formatter/debezium_json.rs       |   2 +-
 src/connector/src/sink/remote.rs              |   3 +-
 src/connector/src/sink/sqlserver.rs           |   7 +-
 src/connector/src/sink/starrocks.rs           |   5 +-
 .../source/iceberg/parquet_file_handler.rs    |   2 +-
 src/expr/impl/src/aggregate/general.rs        |   1 +
 src/expr/impl/src/scalar/cmp.rs               |  25 ++++
 src/expr/impl/src/scalar/date_trunc.rs        |   7 +-
 src/expr/impl/src/scalar/extract.rs           |  42 ++++--
 src/expr/impl/src/scalar/make_time.rs         |  10 +-
 src/expr/impl/src/scalar/timestamptz.rs       |   8 +-
 src/expr/impl/src/scalar/to_char.rs           |   8 +-
 src/expr/impl/src/scalar/to_jsonb.rs          |   7 +-
 src/expr/impl/src/udf/wasm.rs                 |   2 +-
 src/expr/macro/src/lib.rs                     |   2 +-
 src/expr/macro/src/types.rs                   |   4 +-
 src/frontend/src/binder/expr/mod.rs           |   2 +-
 src/frontend/src/expr/literal.rs              |   2 +-
 src/frontend/src/expr/session_timezone.rs     |   6 +-
 src/frontend/src/expr/type_inference/cast.rs  |  36 ++---
 .../src/optimizer/property/monotonicity.rs    |   3 +-
 .../optimizer/rule/index_selection_rule.rs    |   5 +-
 src/frontend/src/utils/data_type.rs           |   4 +-
 src/tests/sqlsmith/src/sql_gen/types.rs       |   2 +-
 59 files changed, 323 insertions(+), 284 deletions(-)

diff --git a/proto/data.proto b/proto/data.proto
index 478fc3e522c8f..5335aed4900fa 100644
--- a/proto/data.proto
+++ b/proto/data.proto
@@ -53,7 +53,7 @@ message DataType {
     SERIAL = 19;
     INT256 = 20;
     MAP = 21;
-    TIMESTAMP_NANOSECOND = 22;
+    Timestamp_ns = 22;
   }
   TypeName type_name = 1;
   // Data length for char.
@@ -105,7 +105,7 @@ enum ArrayType {
   SERIAL = 17;
   INT256 = 18;
   MAP = 20;
-  TIMESTAMP_NANOSECOND = 21;
+  Timestamp_ns = 21;
 }
 
 message Array {
diff --git a/src/batch/executors/src/executor/postgres_query.rs b/src/batch/executors/src/executor/postgres_query.rs
index 9a265a195f054..e7ce698e1e841 100644
--- a/src/batch/executors/src/executor/postgres_query.rs
+++ b/src/batch/executors/src/executor/postgres_query.rs
@@ -85,7 +85,6 @@ 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 10f0041e24eaa..172172bd97208 100644
--- a/src/common/src/array/arrow/arrow_iceberg.rs
+++ b/src/common/src/array/arrow/arrow_iceberg.rs
@@ -97,7 +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::TimestampNs => self.timestamp_ns_type_to_arrow(),
             DataType::Timestamptz => self.timestamptz_type_to_arrow(),
             DataType::Interval => self.interval_type_to_arrow(),
             DataType::Varchar => self.varchar_type_to_arrow(),
@@ -247,7 +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::TimestampNs => self.timestamp_ns_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 199b8dfd48c5c..ba36dea80dd47 100644
--- a/src/common/src/array/arrow/arrow_impl.rs
+++ b/src/common/src/array/arrow/arrow_impl.rs
@@ -107,7 +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::TimestampNs(array) => self.timestamp_ns_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),
@@ -182,9 +182,9 @@ pub trait ToArrow {
     }
 
     #[inline]
-    fn timestampns_to_arrow(
+    fn timestamp_ns_to_arrow(
         &self,
-        array: &TimestampNanosecondArray,
+        array: &TimestampNsArray,
     ) -> Result<arrow_array::ArrayRef, ArrayError> {
         Ok(Arc::new(arrow_array::TimestampNanosecondArray::from(array)))
     }
@@ -328,7 +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::TimestampNs => self.timestamp_ns_type_to_arrow(),
             DataType::Timestamptz => self.timestamptz_type_to_arrow(),
             DataType::Interval => self.interval_type_to_arrow(),
             DataType::Varchar => self.varchar_type_to_arrow(),
@@ -393,7 +393,7 @@ pub trait ToArrow {
         arrow_schema::DataType::Timestamp(arrow_schema::TimeUnit::Microsecond, None)
     }
     #[inline]
-    fn timestampns_type_to_arrow(&self) -> arrow_schema::DataType {
+    fn timestamp_ns_type_to_arrow(&self) -> arrow_schema::DataType {
         arrow_schema::DataType::Timestamp(arrow_schema::TimeUnit::Nanosecond, None)
     }
 
@@ -537,7 +537,7 @@ pub trait FromArrow {
             Timestamp(Second, Some(_)) => DataType::Timestamptz,
             Timestamp(Millisecond, None) => DataType::Timestamp,
             Timestamp(Millisecond, Some(_)) => DataType::Timestamptz,
-            Timestamp(Nanosecond, None) => DataType::TimestampNanosecond,
+            Timestamp(Nanosecond, None) => DataType::TimestampNs,
             Timestamp(Nanosecond, Some(_)) => DataType::Timestamptz,
             Interval(MonthDayNano) => DataType::Interval,
             Utf8 => DataType::Varchar,
@@ -636,10 +636,10 @@ pub trait FromArrow {
                 self.from_timestampus_some_array(array.as_any().downcast_ref().unwrap())
             }
             Timestamp(Nanosecond, None) => {
-                self.from_timestampns_array(array.as_any().downcast_ref().unwrap())
+                self.from_timestamp_ns_array(array.as_any().downcast_ref().unwrap())
             }
             Timestamp(Nanosecond, Some(_)) => {
-                self.from_timestampns_some_array(array.as_any().downcast_ref().unwrap())
+                self.from_timestamp_ns_some_array(array.as_any().downcast_ref().unwrap())
             }
             Interval(MonthDayNano) => {
                 self.from_interval_array(array.as_any().downcast_ref().unwrap())
@@ -811,14 +811,14 @@ pub trait FromArrow {
         Ok(ArrayImpl::Timestamptz(array.into()))
     }
 
-    fn from_timestampns_array(
+    fn from_timestamp_ns_array(
         &self,
         array: &arrow_array::TimestampNanosecondArray,
     ) -> Result<ArrayImpl, ArrayError> {
-        Ok(ArrayImpl::TimestampNanosecond(array.into()))
+        Ok(ArrayImpl::TimestampNs(array.into()))
     }
 
-    fn from_timestampns_some_array(
+    fn from_timestamp_ns_some_array(
         &self,
         array: &arrow_array::TimestampNanosecondArray,
     ) -> Result<ArrayImpl, ArrayError> {
@@ -1060,10 +1060,10 @@ 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_with_timeunit!(TimestampNsArray, arrow_array::TimestampSecondArray, TimeUnit::Second, @map);
+converts_with_timeunit!(TimestampNsArray, arrow_array::TimestampMillisecondArray, TimeUnit::Millisecond, @map);
+converts_with_timeunit!(TimestampNsArray, arrow_array::TimestampMicrosecondArray, TimeUnit::Microsecond, @map);
+converts_with_timeunit!(TimestampNsArray, arrow_array::TimestampNanosecondArray, TimeUnit::Nanosecond, @map);
 
 /// Converts RisingWave value from and into Arrow value.
 trait FromIntoArrow {
@@ -1181,24 +1181,22 @@ impl FromIntoArrowWithUnit for Timestamp {
     }
 }
 
-impl FromIntoArrowWithUnit for TimestampNanosecond {
+impl FromIntoArrowWithUnit for TimestampNs {
     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())
+                TimestampNs(DateTime::from_timestamp(value as _, 0).unwrap().naive_utc())
             }
             TimeUnit::Millisecond => {
-                TimestampNanosecond(DateTime::from_timestamp_millis(value).unwrap().naive_utc())
+                TimestampNs(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())
+                TimestampNs(DateTime::from_timestamp_micros(value).unwrap().naive_utc())
             }
+            TimeUnit::Nanosecond => TimestampNs(DateTime::from_timestamp_nanos(value).naive_utc()),
         }
     }
 
diff --git a/src/common/src/array/chrono_array.rs b/src/common/src/array/chrono_array.rs
index 91fb84d4feb0b..91399fb62d15c 100644
--- a/src/common/src/array/chrono_array.rs
+++ b/src/common/src/array/chrono_array.rs
@@ -12,19 +12,19 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use super::{PrimitiveArray, PrimitiveArrayBuilder, TimestampNanosecond};
+use super::{PrimitiveArray, PrimitiveArrayBuilder, TimestampNs};
 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 TimestampNsArray = PrimitiveArray<TimestampNs>;
 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 TimestampNsArrayBuilder = PrimitiveArrayBuilder<TimestampNs>;
 pub type TimestamptzArrayBuilder = PrimitiveArrayBuilder<Timestamptz>;
 
 #[cfg(test)]
diff --git a/src/common/src/array/mod.rs b/src/common/src/array/mod.rs
index 51e823ac8a5e7..9199781ee2f77 100644
--- a/src/common/src/array/mod.rs
+++ b/src/common/src/array/mod.rs
@@ -45,8 +45,8 @@ pub use bool_array::{BoolArray, BoolArrayBuilder};
 pub use bytes_array::*;
 pub use chrono_array::{
     DateArray, DateArrayBuilder, TimeArray, TimeArrayBuilder, TimestampArray,
-    TimestampArrayBuilder, TimestampNanosecondArray, TimestampNanosecondArrayBuilder,
-    TimestamptzArray, TimestamptzArrayBuilder,
+    TimestampArrayBuilder, TimestampNsArray, TimestampNsArrayBuilder, 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 cfbe539652785..1bab5d2fbbad1 100644
--- a/src/common/src/array/primitive_array.rs
+++ b/src/common/src/array/primitive_array.rs
@@ -129,7 +129,7 @@ impl_primitive_for_others! {
     { Date, Date, Date },
     { Time, Time, Time },
     { Timestamp, Timestamp, Timestamp },
-    { TimestampNanosecond, TimestampNanosecond, TimestampNanosecond },
+    { TimestampNs, TimestampNs, TimestampNs },
     { Timestamptz, Timestamptz, Timestamptz }
 }
 
diff --git a/src/common/src/array/proto_reader.rs b/src/common/src/array/proto_reader.rs
index 19b142cf606a2..e403272384f2c 100644
--- a/src/common/src/array/proto_reader.rs
+++ b/src/common/src/array/proto_reader.rs
@@ -36,9 +36,7 @@ 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::TimestampNs => read_primitive_array::<TimestampNs>(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 d8d9e29b319dc..f09055d790705 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, TimestampNanosecond, Timestamptz, F32, F64,
+    Time, Timestamp, TimestampNs, Timestamptz, F32, F64,
 };
 use crate::util::hash_util::{Crc32FastBuilder, XxHash64Builder};
 use crate::util::sort_util::OrderType;
@@ -584,7 +584,7 @@ impl HashKeyDe for Timestamp {
     }
 }
 
-impl HashKeySer<'_> for TimestampNanosecond {
+impl HashKeySer<'_> for TimestampNs {
     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());
@@ -595,11 +595,11 @@ impl HashKeySer<'_> for TimestampNanosecond {
     }
 }
 
-impl HashKeyDe for TimestampNanosecond {
+impl HashKeyDe for TimestampNs {
     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()
+        TimestampNs::with_secs_nsecs(secs, nsecs).unwrap()
     }
 }
 
diff --git a/src/common/src/row/owned_row.rs b/src/common/src/row/owned_row.rs
index 88c4fc22501e6..9a8f09dd52c5f 100644
--- a/src/common/src/row/owned_row.rs
+++ b/src/common/src/row/owned_row.rs
@@ -18,8 +18,8 @@ use risingwave_common_estimate_size::EstimateSize;
 
 use super::Row;
 use crate::types::{
-    DataType, Date, Datum, DatumRef, Decimal, Interval, ScalarImpl, Time, Timestamp,
-    TimestampNanosecond, ToDatumRef,
+    DataType, Date, Datum, DatumRef, Decimal, Interval, ScalarImpl, Time, Timestamp, TimestampNs,
+    ToDatumRef,
 };
 use crate::util::iter_util::ZipEqDebug;
 use crate::util::value_encoding;
@@ -82,9 +82,7 @@ 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::TimestampNs => x.parse::<TimestampNs>().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 6228379ca7f4d..77f4b41bb46c0 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, TimestampNanosecond, Timestamptz,
+    Timestamp, TimestampNs, Timestamptz,
 };
 
 pub trait RandValue {
@@ -106,9 +106,9 @@ impl RandValue for Timestamp {
     }
 }
 
-impl RandValue for TimestampNanosecond {
+impl RandValue for TimestampNs {
     fn rand_value<R: Rng>(rand: &mut R) -> Self {
-        TimestampNanosecond::new(Date::rand_value(rand).0.and_time(Time::rand_value(rand).0))
+        TimestampNs::new(Date::rand_value(rand).0.and_time(Time::rand_value(rand).0))
     }
 }
 
diff --git a/src/common/src/test_utils/rand_chunk.rs b/src/common/src/test_utils/rand_chunk.rs
index ab01f182df436..341aad3dcc56a 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,
-    TimestampNanosecondArray, TimestamptzArray, Utf8Array,
+    I64Array, Int256Array, IntervalArray, SerialArray, TimeArray, TimestampArray, TimestampNsArray,
+    TimestamptzArray, Utf8Array,
 };
 use crate::test_utils::rand_array::seed_rand_array_ref;
 use crate::types::DataType;
@@ -38,8 +38,8 @@ 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::TimestampNs => {
+                seed_rand_array_ref::<TimestampNsArray>(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 d6c77c616408c..91b25bf8d31c7 100644
--- a/src/common/src/types/datetime.rs
+++ b/src/common/src/types/datetime.rs
@@ -106,7 +106,7 @@ macro_rules! impl_chrono_wrapper {
         }
     };
 }
-impl_chrono_wrapper_without_pgsql_type!(TimestampNanosecond, NaiveDateTime);
+impl_chrono_wrapper_without_pgsql_type!(TimestampNs, NaiveDateTime);
 
 impl_chrono_wrapper!(Date, NaiveDate, DATE);
 impl_chrono_wrapper!(Timestamp, NaiveDateTime, TIMESTAMP);
@@ -198,26 +198,26 @@ impl FromStr for Timestamp {
     }
 }
 
-/// Parse a timestampns from varchar.
+/// Parse a `timestamp_ns` from varchar.
 ///
 /// # Example
 /// ```
 /// use std::str::FromStr;
 ///
-/// use risingwave_common::types::TimestampNanosecond;
+/// use risingwave_common::types::TimestampNs;
 ///
-/// 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();
+/// TimestampNs::from_str("1999-01-08 04:02").unwrap();
+/// TimestampNs::from_str("1999-01-08 04:05:06").unwrap();
+/// TimestampNs::from_str("1999-01-08T04:05:06").unwrap();
+/// TimestampNs::from_str("1999-01-08T04:05:06.123456789").unwrap();
 /// ```
-impl FromStr for TimestampNanosecond {
+impl FromStr for TimestampNs {
     type Err = InvalidParamsError;
 
     fn from_str(s: &str) -> Result<Self> {
         let dt = s
             .parse::<jiff::civil::DateTime>()
-            .map_err(|_| ErrorKind::ParseTimestampNanosecond)?;
+            .map_err(|_| ErrorKind::ParseTimestampNs)?;
         Ok(Self::new(
             Date::from_ymd_uncheck(dt.year() as i32, dt.month() as u32, dt.day() as u32)
                 .0
@@ -256,14 +256,14 @@ impl From<Timestamp> for Date {
 /// ```
 /// use std::str::FromStr;
 ///
-/// use risingwave_common::types::{Date, TimestampNanosecond};
+/// use risingwave_common::types::{Date, TimestampNs};
 ///
-/// let ts = TimestampNanosecond::from_str("1999-01-08 04:02").unwrap();
+/// let ts = TimestampNs::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 {
+impl From<TimestampNs> for Date {
+    fn from(ts: TimestampNs) -> Self {
         Date::new(ts.0.date())
     }
 }
@@ -290,14 +290,14 @@ impl From<Timestamp> for Time {
 /// ```
 /// use std::str::FromStr;
 ///
-/// use risingwave_common::types::{Time, TimestampNanosecond};
+/// use risingwave_common::types::{Time, TimestampNs};
 ///
-/// let ts = TimestampNanosecond::from_str("1999-01-08 04:02").unwrap();
+/// let ts = TimestampNs::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 {
+impl From<TimestampNs> for Time {
+    fn from(ts: TimestampNs) -> Self {
         Time::new(ts.0.time())
     }
 }
@@ -341,8 +341,8 @@ enum ErrorKind {
     ParseTime,
     #[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,
+    #[error("Can't cast string to timestamp_ns (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)")]
+    ParseTimestampNs,
 }
 
 #[derive(Debug, Error)]
@@ -593,34 +593,6 @@ macro_rules! impl_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!(),
-                }
-            }
-        }
-
         impl $type {
             pub fn with_secs_nsecs(secs: i64, nsecs: u32) -> Result<Self> {
                 Ok(Self::new({
@@ -860,30 +832,76 @@ macro_rules! impl_timestamp {
     };
 }
 impl_timestamp!(Timestamp);
-impl_timestamp!(TimestampNanosecond);
+impl_timestamp!(TimestampNs);
+
+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
+        )
+    }
 
-impl From<Timestamp> for TimestampNanosecond {
+    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 ToText for TimestampNs {
+    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::TimestampNs => self.write(f),
+            _ => unreachable!(),
+        }
+    }
+}
+
+impl From<Timestamp> for TimestampNs {
     fn from(ts: Timestamp) -> Self {
-        TimestampNanosecond::new(ts.0)
+        TimestampNs::new(ts.0)
     }
 }
 
-impl From<TimestampNanosecond> for Timestamp {
-    fn from(ts: TimestampNanosecond) -> Self {
+impl From<TimestampNs> for Timestamp {
+    fn from(ts: TimestampNs) -> Self {
         let ts = ts.truncate_micros();
         Timestamp::new(ts.0)
     }
 }
 
-impl TimestampNanosecond {
-    pub fn from_protobuf(cur: &mut Cursor<&[u8]>) -> ArrayResult<TimestampNanosecond> {
+impl TimestampNs {
+    pub fn from_protobuf(cur: &mut Cursor<&[u8]>) -> ArrayResult<TimestampNs> {
         let secs = cur
             .read_i64::<BigEndian>()
-            .context("failed to read i64 from TimestampNanosecond buffer")?;
+            .context("failed to read i64 from TimestampNs buffer")?;
         let nsecs = cur
             .read_u32::<BigEndian>()
-            .context("failed to read u32 from TimestampNanosecond buffer")?;
-        Ok(TimestampNanosecond::with_secs_nsecs(secs, nsecs)?)
+            .context("failed to read u32 from TimestampNs buffer")?;
+        Ok(TimestampNs::with_secs_nsecs(secs, nsecs)?)
     }
 
     pub fn to_protobuf<T: Write>(self, output: &mut T) -> ArrayResult<usize> {
@@ -915,7 +933,7 @@ impl Timestamp {
             .map_err(Into::into)
     }
 
-    pub fn get_timestamp_nanos(&self) -> i64 {
+    pub fn get_timestamp_ns(&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 323b1b7ab66c1..bd418c4f52ff6 100644
--- a/src/common/src/types/macros.rs
+++ b/src/common/src/types/macros.rs
@@ -53,7 +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    },
+            { TimestampNs,    TimestampNs,    timestamp_ns,    $crate::types::TimestampNs,   $crate::types::TimestampNs,           $crate::array::TimestampNsArray,      $crate::array::TimestampNsArrayBuilder    },
             { 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 b75921fe0d121..fbf509bfc17b8 100644
--- a/src/common/src/types/map_type.rs
+++ b/src/common/src/types/map_type.rs
@@ -111,7 +111,7 @@ impl MapType {
             | DataType::Date
             | DataType::Time
             | DataType::Timestamp
-            | DataType::TimestampNanosecond
+            | DataType::TimestampNs
             | DataType::Timestamptz
             | DataType::Interval
             | DataType::Struct(_)
diff --git a/src/common/src/types/mod.rs b/src/common/src/types/mod.rs
index 97b1a50bae304..86f21ec17d9ef 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, TimestampNanosecond};
+pub use self::datetime::{Date, Time, Timestamp, TimestampNs};
 pub use self::decimal::{Decimal, PowError as DecimalPowError};
 pub use self::interval::{test_utils, DateTimeField, Interval, IntervalDisplay};
 pub use self::jsonb::{JsonbRef, JsonbVal};
@@ -151,8 +151,8 @@ pub enum DataType {
     #[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,
+    #[from_str(regex = "(?i)^timestamp_ns$")]
+    TimestampNs,
     #[display("timestamp with time zone")]
     #[from_str(regex = "(?i)^timestamptz$|^timestamp with time zone$")]
     Timestamptz,
@@ -211,7 +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::TimestampNs => Ok(DataType::TimestampNs),
             DataTypeName::Timestamptz => Ok(DataType::Timestamptz),
             DataTypeName::Time => Ok(DataType::Time),
             DataTypeName::Interval => Ok(DataType::Interval),
@@ -238,7 +238,7 @@ impl From<&PbDataType> for DataType {
             PbTypeName::Date => DataType::Date,
             PbTypeName::Time => DataType::Time,
             PbTypeName::Timestamp => DataType::Timestamp,
-            PbTypeName::TimestampNanosecond => DataType::TimestampNanosecond,
+            PbTypeName::TimestampNs => DataType::TimestampNs,
             PbTypeName::Timestamptz => DataType::Timestamptz,
             PbTypeName::Decimal => DataType::Decimal,
             PbTypeName::Interval => DataType::Interval,
@@ -291,7 +291,7 @@ impl From<DataTypeName> for PbTypeName {
             DataTypeName::List => PbTypeName::List,
             DataTypeName::Int256 => PbTypeName::Int256,
             DataTypeName::Map => PbTypeName::Map,
-            DataTypeName::TimestampNanosecond => PbTypeName::TimestampNanosecond,
+            DataTypeName::TimestampNs => PbTypeName::TimestampNs,
         }
     }
 }
@@ -319,7 +319,7 @@ pub mod data_types {
             DataType::Date
                 | DataType::Time
                 | DataType::Timestamp
-                | DataType::TimestampNanosecond
+                | DataType::TimestampNs
                 | DataType::Timestamptz
                 | DataType::Interval
         };
@@ -369,7 +369,7 @@ impl DataType {
             | DataType::Varchar
             | DataType::Time
             | DataType::Timestamp
-            | DataType::TimestampNanosecond
+            | DataType::TimestampNs
             | DataType::Timestamptz
             | DataType::Interval
             | DataType::Bytea
@@ -414,7 +414,7 @@ impl DataType {
         match input {
             DataType::Timestamptz => Some(DataType::Timestamptz),
             DataType::Timestamp | DataType::Date => Some(DataType::Timestamp),
-            DataType::TimestampNanosecond => Some(DataType::TimestampNanosecond),
+            DataType::TimestampNs => Some(DataType::TimestampNs),
             _ => None,
         }
     }
@@ -899,10 +899,7 @@ 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::TimestampNanosecond => {
+            DataType::Struct(_) | DataType::List(_) | DataType::Map(_) | DataType::TimestampNs => {
                 return Err(format!("unsupported data type: {}", data_type).into());
             }
         };
@@ -926,7 +923,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(),
+            DataType::TimestampNs => TimestampNs::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(),
@@ -1032,7 +1029,7 @@ impl ScalarRefImpl<'_> {
                 v.0.and_utc().timestamp().serialize(&mut *ser)?;
                 v.0.and_utc().timestamp_subsec_nanos().serialize(ser)?;
             }
-            Self::TimestampNanosecond(v) => {
+            Self::TimestampNs(v) => {
                 v.0.and_utc().timestamp().serialize(&mut *ser)?;
                 v.0.and_utc().timestamp_subsec_nanos().serialize(ser)?;
             }
@@ -1091,10 +1088,10 @@ impl ScalarImpl {
                 Timestamp::with_secs_nsecs(secs, nsecs)
                     .map_err(|e| memcomparable::Error::Message(e.to_report_string()))?
             }),
-            Ty::TimestampNanosecond => Self::TimestampNanosecond({
+            Ty::TimestampNs => Self::TimestampNs({
                 let secs = i64::deserialize(&mut *de)?;
                 let nsecs = u32::deserialize(de)?;
-                TimestampNanosecond::with_secs_nsecs(secs, nsecs)
+                TimestampNs::with_secs_nsecs(secs, nsecs)
                     .map_err(|e| memcomparable::Error::Message(e.to_report_string()))?
             }),
             Ty::Timestamptz => Self::Timestamptz(Timestamptz::deserialize(de)?),
@@ -1261,11 +1258,9 @@ 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::TimestampNs => (
+                    ScalarImpl::TimestampNs(TimestampNs::from_timestamp_uncheck(23333333, 2333)),
+                    DataType::TimestampNs,
                 ),
                 DataTypeName::Timestamptz => (
                     ScalarImpl::Timestamptz(Timestamptz::from_micros(233333333)),
diff --git a/src/common/src/types/postgres_type.rs b/src/common/src/types/postgres_type.rs
index beb718cfbe1d3..1980d5e708edb 100644
--- a/src/common/src/types/postgres_type.rs
+++ b/src/common/src/types/postgres_type.rs
@@ -69,7 +69,7 @@ impl DataType {
                     $(
                     DataType::$enum => $len,
                     )*
-                    DataType::TimestampNanosecond => 8,
+                    DataType::TimestampNs => 8,
                     DataType::Serial => 8,
                     DataType::Int256 => -1,
                     DataType::List(_) | DataType::Struct(_) | DataType::Map(_) => -1,
@@ -122,13 +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::TimestampNs => 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,
+                    DataType::TimestampNs => 1305,
                     // TODO: Support to give a new oid for custom struct type. #9434
                     DataType::Struct(_) => 2249,  // pseudo-type of struct (see `pg_type.dat`)
                 }
@@ -148,7 +148,7 @@ impl DataType {
                     DataType::List(_) => "list",
                     DataType::Serial => "serial",
                     DataType::Int256 => "rw_int256",
-                    DataType::TimestampNanosecond => "timestamp_nano",
+                    DataType::TimestampNs => "timestamp_ns",
                     DataType::Map(_) => "map",
                 }
             }
diff --git a/src/common/src/types/scalar_impl.rs b/src/common/src/types/scalar_impl.rs
index cad9fcc7aae40..5ca296e0fcbb9 100644
--- a/src/common/src/types/scalar_impl.rs
+++ b/src/common/src/types/scalar_impl.rs
@@ -246,19 +246,19 @@ impl ScalarRef<'_> for Timestamp {
 }
 
 /// Implement `Scalar` for `Timestamp`.
-impl Scalar for TimestampNanosecond {
-    type ScalarRefType<'a> = TimestampNanosecond;
+impl Scalar for TimestampNs {
+    type ScalarRefType<'a> = TimestampNs;
 
-    fn as_scalar_ref(&self) -> TimestampNanosecond {
+    fn as_scalar_ref(&self) -> TimestampNs {
         *self
     }
 }
 
 /// Implement `ScalarRef` for `Timestamp`.
-impl ScalarRef<'_> for TimestampNanosecond {
-    type ScalarType = TimestampNanosecond;
+impl ScalarRef<'_> for TimestampNs {
+    type ScalarType = TimestampNs;
 
-    fn to_owned_scalar(&self) -> TimestampNanosecond {
+    fn to_owned_scalar(&self) -> TimestampNs {
         *self
     }
 
diff --git a/src/common/src/types/to_binary.rs b/src/common/src/types/to_binary.rs
index 0b3933e9c8b37..baa03a0ef0af8 100644
--- a/src/common/src/types/to_binary.rs
+++ b/src/common/src/types/to_binary.rs
@@ -17,7 +17,7 @@ use postgres_types::{ToSql, Type};
 use rw_iter_util::ZipEqFast;
 
 use super::{
-    DataType, Date, Decimal, Interval, ScalarRefImpl, Serial, Time, Timestamp, TimestampNanosecond,
+    DataType, Date, Decimal, Interval, ScalarRefImpl, Serial, Time, Timestamp, TimestampNs,
     Timestamptz, F32, F64,
 };
 use crate::array::{ListRef, StructRef};
@@ -74,7 +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 },
+    { TimestampNs, TimestampNs, |x: &TimestampNs| x.0 },
     { Decimal, Decimal, |x| x },
     { Interval, Interval, |x| x },
     { Serial, Serial, |x: &Serial| x.0 },
@@ -157,7 +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::TimestampNs(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 c5a5254c34746..734cbca8fd510 100644
--- a/src/common/src/types/to_sql.rs
+++ b/src/common/src/types/to_sql.rs
@@ -47,7 +47,7 @@ impl ToSql for ScalarImpl {
             ScalarImpl::Int256(_)
             | ScalarImpl::Struct(_)
             | ScalarImpl::List(_)
-            | ScalarImpl::TimestampNanosecond(_) => {
+            | ScalarImpl::TimestampNs(_) => {
                 bail_not_implemented!("the postgres encoding for {ty} is unsupported")
             }
             ScalarImpl::Map(_) => todo!(),
@@ -90,7 +90,7 @@ impl ToSql for ScalarRefImpl<'_> {
             ScalarRefImpl::Int256(_)
             | ScalarRefImpl::Struct(_)
             | ScalarRefImpl::List(_)
-            | ScalarRefImpl::TimestampNanosecond(_) => {
+            | ScalarRefImpl::TimestampNs(_) => {
                 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 17f7207b12758..65e660783e6b9 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, TimestampNanosecond, Timestamptz, F32, F64,
+    Time, Timestamp, TimestampNs, Timestamptz, F32, F64,
 };
 
 /// A trait for all physical types that can be associated with a [`DataType`].
@@ -103,7 +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!(TimestampNs, DataType::TimestampNs);
 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 eb64740285484..fd6faeb1b2980 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, TimestampNanosecond,
-    Timestamptz, ToDatumRef, F32, F64,
+    DataType, Date, Datum, Int256, ScalarImpl, Serial, Time, Timestamp, TimestampNs, Timestamptz,
+    ToDatumRef, F32, F64,
 };
 use crate::util::sort_util::{ColumnOrder, OrderType};
 
@@ -143,7 +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::TimestampNs => size_of::<TimestampNs>(),
             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 d995620f18aee..8dd87a50af022 100644
--- a/src/common/src/util/value_encoding/mod.rs
+++ b/src/common/src/util/value_encoding/mod.rs
@@ -220,7 +220,7 @@ fn serialize_scalar(value: ScalarRefImpl<'_>, buf: &mut impl BufMut) {
             v.0.and_utc().timestamp_subsec_nanos(),
             buf,
         ),
-        ScalarRefImpl::TimestampNanosecond(v) => serialize_timestamp(
+        ScalarRefImpl::TimestampNs(v) => serialize_timestamp(
             v.0.and_utc().timestamp(),
             v.0.and_utc().timestamp_subsec_nanos(),
             buf,
@@ -252,7 +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::TimestampNs(_) => 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
@@ -352,9 +352,7 @@ 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::TimestampNs => ScalarImpl::TimestampNs(deserialize_timestamp_ns(data)?),
         DataType::Timestamptz => {
             ScalarImpl::Timestamptz(Timestamptz::from_micros(data.get_i64_le()))
         }
@@ -443,10 +441,10 @@ 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> {
+fn deserialize_timestamp_ns(data: &mut impl Buf) -> Result<TimestampNs> {
     let secs = data.get_i64_le();
     let nsecs = data.get_u32_le();
-    TimestampNanosecond::with_secs_nsecs(secs, nsecs)
+    TimestampNs::with_secs_nsecs(secs, nsecs)
         .map_err(|_e| ValueEncodingError::InvalidTimestampEncoding(secs, nsecs))
 }
 
diff --git a/src/connector/src/parser/csv_parser.rs b/src/connector/src/parser/csv_parser.rs
index 72ad755acc4c9..831cb8460b4aa 100644
--- a/src/connector/src/parser/csv_parser.rs
+++ b/src/connector/src/parser/csv_parser.rs
@@ -14,7 +14,7 @@
 
 use risingwave_common::cast::str_to_bool;
 use risingwave_common::types::{
-    Date, Decimal, ScalarImpl, Time, Timestamp, TimestampNanosecond, Timestamptz,
+    Date, Decimal, ScalarImpl, Time, Timestamp, TimestampNs, Timestamptz,
 };
 
 use super::unified::{AccessError, AccessResult};
@@ -99,7 +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::TimestampNs => parse!(v, TimestampNs)?.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 a39b1c348c199..bed01cd2de817 100644
--- a/src/connector/src/parser/mysql.rs
+++ b/src/connector/src/parser/mysql.rs
@@ -27,7 +27,7 @@ use anyhow::anyhow;
 use chrono::NaiveDate;
 use risingwave_common::bail;
 use risingwave_common::types::{
-    DataType, Date, Datum, Decimal, JsonbVal, ScalarImpl, Time, Timestamp, Timestamptz,
+    DataType, Date, Datum, Decimal, JsonbVal, ScalarImpl, Time, Timestamp, TimestampNs, Timestamptz,
 };
 use rust_decimal::Decimal as RustDecimal;
 
@@ -154,13 +154,13 @@ pub fn mysql_datum_to_rw_datum(
                 Timestamp
             )
         }
-        DataType::TimestampNanosecond => {
+        DataType::TimestampNs => {
             handle_data_type!(
                 mysql_row,
                 mysql_datum_index,
                 column_name,
                 chrono::NaiveDateTime,
-                Timestamp
+                TimestampNs
             )
         }
         DataType::Timestamptz => {
diff --git a/src/connector/src/parser/postgres.rs b/src/connector/src/parser/postgres.rs
index 598a0d973a4c2..db565408fbccc 100644
--- a/src/connector/src/parser/postgres.rs
+++ b/src/connector/src/parser/postgres.rs
@@ -68,7 +68,6 @@ fn postgres_cell_to_scalar_impl(
         | DataType::Date
         | DataType::Time
         | DataType::Timestamp
-        | DataType::TimestampNanosecond
         | DataType::Timestamptz
         | DataType::Jsonb
         | DataType::Interval
@@ -117,7 +116,7 @@ fn postgres_cell_to_scalar_impl(
                 }
             }
         },
-        DataType::Struct(_) | DataType::Serial | DataType::Map(_) => {
+        DataType::Struct(_) | DataType::Serial | DataType::Map(_) | DataType::TimestampNs => {
             // Is this branch reachable?
             // Struct and Serial are not supported
             tracing::warn!(name, ?data_type, "unsupported data type, set to null");
diff --git a/src/connector/src/parser/unified/json.rs b/src/connector/src/parser/unified/json.rs
index 823f645dd2686..86e20644c717d 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,
-    TimestampNanosecond, Timestamptz, ToOwnedDatum,
+    DataType, Date, Decimal, Int256, Interval, JsonbVal, ScalarImpl, Time, Timestamp, TimestampNs,
+    Timestamptz, ToOwnedDatum,
 };
 use risingwave_common::util::iter_util::ZipEqFast;
 use risingwave_connector_codec::decoder::utils::scaled_bigint_to_rust_decimal;
@@ -487,10 +487,10 @@ impl JsonParseOptions {
             ) => i64_to_timestamp(value.as_i64().unwrap())
                 .map_err(|_| create_error())?
                 .into(),
-            (DataType::TimestampNanosecond, ValueType::String) => value
+            (DataType::TimestampNs, ValueType::String) => value
                 .as_str()
                 .unwrap()
-                .parse::<TimestampNanosecond>()
+                .parse::<TimestampNs>()
                 .map_err(|_| create_error())?
                 .into(),
             // ---- Timestamptz -----
diff --git a/src/connector/src/sink/big_query.rs b/src/connector/src/sink/big_query.rs
index 3f0769180599a..909dc718b3af6 100644
--- a/src/connector/src/sink/big_query.rs
+++ b/src/connector/src/sink/big_query.rs
@@ -317,7 +317,7 @@ impl BigQuerySink {
             DataType::Int32 => Ok("INT64".to_owned()),
             DataType::Int64 => Ok("INT64".to_owned()),
             DataType::Float32 => Err(SinkError::BigQuery(anyhow::anyhow!(
-                "Bigquery cannot support real"
+                "Bigquery don't support Real"
             ))),
             DataType::Float64 => Ok("FLOAT64".to_owned()),
             DataType::Decimal => Ok("NUMERIC".to_owned()),
@@ -325,7 +325,6 @@ 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) => {
@@ -345,10 +344,13 @@ impl BigQuerySink {
             DataType::Jsonb => Ok("JSON".to_owned()),
             DataType::Serial => Ok("INT64".to_owned()),
             DataType::Int256 => Err(SinkError::BigQuery(anyhow::anyhow!(
-                "Bigquery cannot support Int256"
+                "Bigquery don't support Int256"
             ))),
             DataType::Map(_) => Err(SinkError::BigQuery(anyhow::anyhow!(
-                "Bigquery cannot support Map"
+                "Bigquery don't support Map"
+            ))),
+            DataType::TimestampNs => Err(SinkError::BigQuery(anyhow::anyhow!(
+                "Bigquery don't support TimestampNs"
             ))),
         }
     }
@@ -361,7 +363,7 @@ impl BigQuerySink {
             }
             DataType::Float32 => {
                 return Err(SinkError::BigQuery(anyhow::anyhow!(
-                    "Bigquery cannot support real"
+                    "Bigquery don't support Real"
                 )))
             }
             DataType::Float64 => TableFieldSchema::float(&rw_field.name),
@@ -370,11 +372,15 @@ 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::TimestampNs => {
+                return Err(SinkError::BigQuery(anyhow::anyhow!(
+                    "Bigquery don't support TimestampNs"
+                )))
+            }
             DataType::Timestamptz => TableFieldSchema::timestamp(&rw_field.name),
             DataType::Interval => {
                 return Err(SinkError::BigQuery(anyhow::anyhow!(
-                    "Bigquery cannot support Interval"
+                    "Bigquery don't support Interval"
                 )))
             }
             DataType::Struct(_) => {
@@ -397,12 +403,12 @@ impl BigQuerySink {
             DataType::Jsonb => TableFieldSchema::json(&rw_field.name),
             DataType::Int256 => {
                 return Err(SinkError::BigQuery(anyhow::anyhow!(
-                    "Bigquery cannot support Int256"
+                    "Bigquery don't support Int256"
                 )))
             }
             DataType::Map(_) => {
                 return Err(SinkError::BigQuery(anyhow::anyhow!(
-                    "Bigquery cannot support Map"
+                    "Bigquery don't support Map"
                 )))
             }
         };
@@ -896,9 +902,7 @@ 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::TimestampNs => 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 eca3f3ebaf49f..4a676f28f8cdc 100644
--- a/src/connector/src/sink/clickhouse.rs
+++ b/src/connector/src/sink/clickhouse.rs
@@ -448,11 +448,9 @@ impl ClickHouseSink {
                 "clickhouse can not support Time".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::TimestampNs => 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"))
             }
@@ -948,7 +946,7 @@ impl ClickHouseFieldWithNull {
                     "clickhouse can not support Time".to_owned(),
                 ))
             }
-            ScalarRefImpl::Timestamp(_) | ScalarRefImpl::TimestampNanosecond(_) => {
+            ScalarRefImpl::Timestamp(_) | ScalarRefImpl::TimestampNs(_) => {
                 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 b33f79f134b3e..f83a8365fc9da 100644
--- a/src/connector/src/sink/doris.rs
+++ b/src/connector/src/sink/doris.rs
@@ -168,29 +168,32 @@ impl DorisSink {
                 Ok(doris_data_type.contains("STRING") | doris_data_type.contains("VARCHAR"))
             }
             risingwave_common::types::DataType::Time => {
-                Err(SinkError::Doris("doris can not support Time".to_owned()))
+                Err(SinkError::Doris("doris don't support Time".to_owned()))
             }
-            risingwave_common::types::DataType::Timestamp | risingwave_common::types::DataType::TimestampNanosecond => {
+            risingwave_common::types::DataType::Timestamp => {
                 Ok(doris_data_type.contains("DATETIME"))
             }
             risingwave_common::types::DataType::Timestamptz => Err(SinkError::Doris(
                 "TIMESTAMP WITH TIMEZONE is not supported for Doris sink as Doris doesn't store time values with timezone information. Please convert to TIMESTAMP first.".to_owned(),
             )),
             risingwave_common::types::DataType::Interval => Err(SinkError::Doris(
-                "doris can not support Interval".to_owned(),
+                "doris don't support Interval".to_owned(),
             )),
             risingwave_common::types::DataType::Struct(_) => Ok(doris_data_type.contains("STRUCT")),
             risingwave_common::types::DataType::List(_) => Ok(doris_data_type.contains("ARRAY")),
             risingwave_common::types::DataType::Bytea => {
-                Err(SinkError::Doris("doris can not support Bytea".to_owned()))
+                Err(SinkError::Doris("doris don't support Bytea".to_owned()))
             }
             risingwave_common::types::DataType::Jsonb => Ok(doris_data_type.contains("JSON")),
             risingwave_common::types::DataType::Serial => Ok(doris_data_type.contains("BIGINT")),
             risingwave_common::types::DataType::Int256 => {
-                Err(SinkError::Doris("doris can not support Int256".to_owned()))
+                Err(SinkError::Doris("doris don't support Int256".to_owned()))
             }
             risingwave_common::types::DataType::Map(_) => {
-                Err(SinkError::Doris("doris can not support Map".to_owned()))
+                Err(SinkError::Doris("doris don't support Map".to_owned()))
+            }
+            | risingwave_common::types::DataType::TimestampNs => {
+                Err(SinkError::Doris("doris don't support TimestampNs".to_owned()))
             }
         }
     }
diff --git a/src/connector/src/sink/dynamodb.rs b/src/connector/src/sink/dynamodb.rs
index c18f84302d960..4f1f000357d13 100644
--- a/src/connector/src/sink/dynamodb.rs
+++ b/src/connector/src/sink/dynamodb.rs
@@ -319,7 +319,7 @@ fn map_data(scalar_ref: Option<ScalarRefImpl<'_>>, data_type: &DataType) -> Resu
         | DataType::Time
         | DataType::Timestamp
         | DataType::Timestamptz
-        | DataType::TimestampNanosecond
+        | DataType::TimestampNs
         | 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 74070e993a620..29b3f37779772 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 | DataType::TimestampNanosecond => return no_match_err(),
+        DataType::Timestamp | DataType::TimestampNs => 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 77630822a50c9..44bd7595e55ff 100644
--- a/src/connector/src/sink/encoder/bson.rs
+++ b/src/connector/src/sink/encoder/bson.rs
@@ -144,7 +144,7 @@ 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)) => {
+        (DataType::TimestampNs, ScalarRefImpl::TimestampNs(v)) => {
             Bson::DateTime(DateTime::from_millis(v.0.and_utc().timestamp_millis()))
         }
         (DataType::Timestamptz, ScalarRefImpl::Timestamptz(v)) => {
diff --git a/src/connector/src/sink/encoder/json.rs b/src/connector/src/sink/encoder/json.rs
index a7f704da0b172..1cd2364833911 100644
--- a/src/connector/src/sink/encoder/json.rs
+++ b/src/connector/src/sink/encoder/json.rs
@@ -394,7 +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::TimestampNs => "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 bf411d2231a89..fb5510eca8893 100644
--- a/src/connector/src/sink/encoder/proto.rs
+++ b/src/connector/src/sink/encoder/proto.rs
@@ -430,9 +430,9 @@ 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() {
+        DataType::TimestampNs => match proto_field.kind() {
             Kind::String => {
-                maybe.on_base(|s| Ok(Value::String(s.into_timestampns().to_string())))?
+                maybe.on_base(|s| Ok(Value::String(s.into_timestamp_ns().to_string())))?
             }
             _ => return no_match_err(), // google.type.DateTime
         },
diff --git a/src/connector/src/sink/formatter/debezium_json.rs b/src/connector/src/sink/formatter/debezium_json.rs
index 498af2491e080..bb11f9faa32b7 100644
--- a/src/connector/src/sink/formatter/debezium_json.rs
+++ b/src/connector/src/sink/formatter/debezium_json.rs
@@ -299,7 +299,7 @@ pub(crate) fn field_to_json(field: &Field) -> Value {
         // 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::TimestampNanosecond => {
+        | risingwave_common::types::DataType::TimestampNs => {
             ("int64", "org.apache.kafka.connect.data.Timestamp")
         }
 
diff --git a/src/connector/src/sink/remote.rs b/src/connector/src/sink/remote.rs
index 0e4cb1187e3b1..1c38bcf8b7677 100644
--- a/src/connector/src/sink/remote.rs
+++ b/src/connector/src/sink/remote.rs
@@ -187,7 +187,6 @@ 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
@@ -217,7 +216,7 @@ async fn validate_remote_sink(param: &SinkParam, sink_name: &str) -> ConnectorRe
                     )))
                 }
             },
-            DataType::Serial | DataType::Int256 | DataType::Map(_) => Err(SinkError::Remote(anyhow!(
+            DataType::Serial | DataType::Int256 | DataType::Map(_) | DataType::TimestampNs => Err(SinkError::Remote(anyhow!(
                             "remote sink supports Int16, Int32, Int64, Float32, Float64, Boolean, Decimal, Time, Date, Interval, Jsonb, Timestamp, Timestamptz, Bytea, List and Varchar, (Es sink support Struct) got {:?}: {:?}",
                             col.name,
                             col.data_type,
diff --git a/src/connector/src/sink/sqlserver.rs b/src/connector/src/sink/sqlserver.rs
index a41d8944c34cd..eb2152fd1ab20 100644
--- a/src/connector/src/sink/sqlserver.rs
+++ b/src/connector/src/sink/sqlserver.rs
@@ -580,7 +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::TimestampNs(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()),
@@ -620,7 +620,7 @@ fn bind_params(
                 DataType::Time => {
                     query.bind(None as Option<chrono::NaiveTime>);
                 }
-                DataType::Timestamp | DataType::TimestampNanosecond => {
+                DataType::Timestamp => {
                     query.bind(None as Option<chrono::NaiveDateTime>);
                 }
                 DataType::Timestamptz => {
@@ -639,6 +639,7 @@ fn bind_params(
                 DataType::Serial => return Err(data_type_not_supported("Serial")),
                 DataType::Int256 => return Err(data_type_not_supported("Int256")),
                 DataType::Map(_) => return Err(data_type_not_supported("Map")),
+                DataType::TimestampNs => return Err(data_type_not_supported("TimestampNs")),
             },
         };
     }
@@ -664,7 +665,6 @@ 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")),
@@ -674,6 +674,7 @@ fn check_data_type_compatibility(data_type: &DataType) -> Result<()> {
         DataType::Serial => Err(data_type_not_supported("Serial")),
         DataType::Int256 => Err(data_type_not_supported("Int256")),
         DataType::Map(_) => Err(data_type_not_supported("Map")),
+        DataType::TimestampNs => Err(data_type_not_supported("TimestampNs")),
     }
 }
 
diff --git a/src/connector/src/sink/starrocks.rs b/src/connector/src/sink/starrocks.rs
index 831cb4ae59103..2aaf3339547a4 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::TimestampNanosecond => {
+            risingwave_common::types::DataType::Timestamp => {
                 Ok(starrocks_data_type.contains("datetime"))
             }
             risingwave_common::types::DataType::Timestamptz => Err(SinkError::Starrocks(
@@ -251,6 +251,9 @@ impl StarrocksSink {
             risingwave_common::types::DataType::Map(_) => Err(SinkError::Starrocks(
                 "MAP is not supported for Starrocks sink.".to_owned(),
             )),
+            risingwave_common::types::DataType::TimestampNs => Err(SinkError::Starrocks(
+                "TIMESTAMPNS is not supported for Starrocks sink.".to_owned(),
+            )),
         }
     }
 }
diff --git a/src/connector/src/source/iceberg/parquet_file_handler.rs b/src/connector/src/source/iceberg/parquet_file_handler.rs
index 6e455ad09af16..5cdadfe773d28 100644
--- a/src/connector/src/source/iceberg/parquet_file_handler.rs
+++ b/src/connector/src/source/iceberg/parquet_file_handler.rs
@@ -416,7 +416,7 @@ pub fn is_parquet_schema_match_source_schema(
             )
             | (
                 ArrowDateType::Timestamp(TimeUnit::Nanosecond, None),
-                RwDataType::TimestampNanosecond
+                RwDataType::TimestampNs
             )
             | (
                 ArrowDateType::Timestamp(_, Some(_)),
diff --git a/src/expr/impl/src/aggregate/general.rs b/src/expr/impl/src/aggregate/general.rs
index b6f36e69085eb..c21d3bb68616b 100644
--- a/src/expr/impl/src/aggregate/general.rs
+++ b/src/expr/impl/src/aggregate/general.rs
@@ -96,6 +96,7 @@ fn _var_samp() {}
 #[aggregate("min(time) -> auto", state = "ref")]
 #[aggregate("min(interval) -> auto", state = "ref")]
 #[aggregate("min(timestamp) -> auto", state = "ref")]
+#[aggregate("min(timestamp_ns) -> auto", state = "ref")]
 #[aggregate("min(timestamptz) -> auto", state = "ref")]
 #[aggregate("min(varchar) -> auto", state = "ref")]
 #[aggregate("min(bytea) -> auto", state = "ref")]
diff --git a/src/expr/impl/src/scalar/cmp.rs b/src/expr/impl/src/scalar/cmp.rs
index 4fe852550c15e..c8d4a5097ce94 100644
--- a/src/expr/impl/src/scalar/cmp.rs
+++ b/src/expr/impl/src/scalar/cmp.rs
@@ -31,9 +31,12 @@ use risingwave_expr::function;
 #[function("equal(time, time) -> boolean")]
 #[function("equal(interval, interval) -> boolean")]
 #[function("equal(timestamp, timestamp) -> boolean")]
+#[function("equal(timestamp_ns, timestamp_ns) -> boolean")]
 #[function("equal(timestamptz, timestamptz) -> boolean")]
 #[function("equal(date, timestamp) -> boolean")]
+#[function("equal(date, timestamp_ns) -> boolean")]
 #[function("equal(timestamp, date) -> boolean")]
+#[function("equal(timestamp_ns, date) -> boolean")]
 #[function("equal(time, interval) -> boolean")]
 #[function("equal(interval, time) -> boolean")]
 #[function("equal(varchar, varchar) -> boolean")]
@@ -60,9 +63,12 @@ where
 #[function("not_equal(time, time) -> boolean")]
 #[function("not_equal(interval, interval) -> boolean")]
 #[function("not_equal(timestamp, timestamp) -> boolean")]
+#[function("not_equal(timestamp_ns, timestamp_ns) -> boolean")]
 #[function("not_equal(timestamptz, timestamptz) -> boolean")]
 #[function("not_equal(date, timestamp) -> boolean")]
+#[function("not_equal(date, timestamp_ns) -> boolean")]
 #[function("not_equal(timestamp, date) -> boolean")]
+#[function("not_equal(timestamp_ns, date) -> boolean")]
 #[function("not_equal(time, interval) -> boolean")]
 #[function("not_equal(interval, time) -> boolean")]
 #[function("not_equal(varchar, varchar) -> boolean")]
@@ -95,7 +101,9 @@ where
 #[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(date, timestamp_ns) -> boolean")]
 #[function("greater_than_or_equal(timestamp, date) -> boolean")]
+#[function("greater_than_or_equal(timestamp_ns, date) -> boolean")]
 #[function("greater_than_or_equal(time, interval) -> boolean")]
 #[function("greater_than_or_equal(interval, time) -> boolean")]
 #[function("greater_than_or_equal(varchar, varchar) -> boolean")]
@@ -121,9 +129,12 @@ where
 #[function("greater_than(time, time) -> boolean")]
 #[function("greater_than(interval, interval) -> boolean")]
 #[function("greater_than(timestamp, timestamp) -> boolean")]
+#[function("greater_than(timestamp_ns, timestamp_ns) -> boolean")]
 #[function("greater_than(timestamptz, timestamptz) -> boolean")]
 #[function("greater_than(date, timestamp) -> boolean")]
+#[function("greater_than(date, timestamp_ns) -> boolean")]
 #[function("greater_than(timestamp, date) -> boolean")]
+#[function("greater_than(timestamp_ns, date) -> boolean")]
 #[function("greater_than(time, interval) -> boolean")]
 #[function("greater_than(interval, time) -> boolean")]
 #[function("greater_than(varchar, varchar) -> boolean")]
@@ -152,9 +163,12 @@ where
 #[function("less_than_or_equal(time, time) -> boolean")]
 #[function("less_than_or_equal(interval, interval) -> boolean")]
 #[function("less_than_or_equal(timestamp, timestamp) -> boolean")]
+#[function("less_than_or_equal(timestamp_ns, timestamp_ns) -> boolean")]
 #[function("less_than_or_equal(timestamptz, timestamptz) -> boolean")]
 #[function("less_than_or_equal(date, timestamp) -> boolean")]
+#[function("less_than_or_equal(date, timestamp_ns) -> boolean")]
 #[function("less_than_or_equal(timestamp, date) -> boolean")]
+#[function("less_than_or_equal(timestamp_ns, date) -> boolean")]
 #[function("less_than_or_equal(time, interval) -> boolean")]
 #[function("less_than_or_equal(interval, time) -> boolean")]
 #[function("less_than_or_equal(varchar, varchar) -> boolean")]
@@ -180,9 +194,12 @@ where
 #[function("less_than(time, time) -> boolean")]
 #[function("less_than(interval, interval) -> boolean")]
 #[function("less_than(timestamp, timestamp) -> boolean")]
+#[function("less_than(timestamp_ns, timestamp_ns) -> boolean")]
 #[function("less_than(timestamptz, timestamptz) -> boolean")]
 #[function("less_than(date, timestamp) -> boolean")]
+#[function("less_than(date, timestamp_ns) -> boolean")]
 #[function("less_than(timestamp, date) -> boolean")]
+#[function("less_than(timestamp_ns, date) -> boolean")]
 #[function("less_than(time, interval) -> boolean")]
 #[function("less_than(interval, time) -> boolean")]
 #[function("less_than(varchar, varchar) -> boolean")]
@@ -211,9 +228,12 @@ where
 #[function("is_distinct_from(time, time) -> boolean")]
 #[function("is_distinct_from(interval, interval) -> boolean")]
 #[function("is_distinct_from(timestamp, timestamp) -> boolean")]
+#[function("is_distinct_from(timestamp_ns, timestamp_ns) -> boolean")]
 #[function("is_distinct_from(timestamptz, timestamptz) -> boolean")]
 #[function("is_distinct_from(date, timestamp) -> boolean")]
+#[function("is_distinct_from(date, timestamp_ns) -> boolean")]
 #[function("is_distinct_from(timestamp, date) -> boolean")]
+#[function("is_distinct_from(timestamp_ns, date) -> boolean")]
 #[function("is_distinct_from(time, interval) -> boolean")]
 #[function("is_distinct_from(interval, time) -> boolean")]
 #[function("is_distinct_from(varchar, varchar) -> boolean")]
@@ -242,9 +262,12 @@ where
 #[function("is_not_distinct_from(time, time) -> boolean")]
 #[function("is_not_distinct_from(interval, interval) -> boolean")]
 #[function("is_not_distinct_from(timestamp, timestamp) -> boolean")]
+#[function("is_not_distinct_from(timestamp_ns, timestamp_ns) -> boolean")]
 #[function("is_not_distinct_from(timestamptz, timestamptz) -> boolean")]
 #[function("is_not_distinct_from(date, timestamp) -> boolean")]
+#[function("is_not_distinct_from(date, timestamp_ns) -> boolean")]
 #[function("is_not_distinct_from(timestamp, date) -> boolean")]
+#[function("is_not_distinct_from(timestamp_ns, date) -> boolean")]
 #[function("is_not_distinct_from(time, interval) -> boolean")]
 #[function("is_not_distinct_from(interval, time) -> boolean")]
 #[function("is_not_distinct_from(varchar, varchar) -> boolean")]
@@ -303,6 +326,7 @@ fn is_not_null<T>(v: Option<T>) -> bool {
 #[function("greatest(...) -> time")]
 #[function("greatest(...) -> interval")]
 #[function("greatest(...) -> timestamp")]
+#[function("greatest(...) -> timestamp_ns")]
 #[function("greatest(...) -> timestamptz")]
 #[function("greatest(...) -> varchar")]
 #[function("greatest(...) -> bytea")]
@@ -333,6 +357,7 @@ where
 #[function("least(...) -> time")]
 #[function("least(...) -> interval")]
 #[function("least(...) -> timestamp")]
+#[function("least(...) -> timestamp_ns")]
 #[function("least(...) -> timestamptz")]
 #[function("least(...) -> varchar")]
 #[function("least(...) -> bytea")]
diff --git a/src/expr/impl/src/scalar/date_trunc.rs b/src/expr/impl/src/scalar/date_trunc.rs
index 0c8d176bf6e33..061ca594e790d 100644
--- a/src/expr/impl/src/scalar/date_trunc.rs
+++ b/src/expr/impl/src/scalar/date_trunc.rs
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use risingwave_common::types::{Interval, Timestamp, TimestampNanosecond, Timestamptz};
+use risingwave_common::types::{Interval, Timestamp, TimestampNs, Timestamptz};
 use risingwave_expr::{function, ExprError, Result};
 
 use super::timestamptz::timestamp_at_time_zone;
@@ -34,10 +34,7 @@ const CENTURY: &str = "century";
 const MILLENNIUM: &str = "millennium";
 
 #[function("date_trunc(varchar, timestamp_ns) -> timestamp_ns")]
-pub fn date_trunc_timestamp_ns(
-    field: &str,
-    ts: TimestampNanosecond,
-) -> Result<TimestampNanosecond> {
+pub fn date_trunc_timestamp_ns(field: &str, ts: TimestampNs) -> Result<TimestampNs> {
     Ok(match field.to_ascii_lowercase().as_str() {
         NANOSECONDS => ts,
         MICROSECONDS => ts.truncate_micros(),
diff --git a/src/expr/impl/src/scalar/extract.rs b/src/expr/impl/src/scalar/extract.rs
index f9aed8b13bd7b..47f50f240eb04 100644
--- a/src/expr/impl/src/scalar/extract.rs
+++ b/src/expr/impl/src/scalar/extract.rs
@@ -16,7 +16,7 @@ use std::str::FromStr;
 
 use chrono::{Datelike, NaiveTime, Timelike};
 use risingwave_common::types::{
-    Date, Decimal, Interval, Time, Timestamp, TimestampNanosecond, Timestamptz, F64,
+    Date, Decimal, Interval, Time, Timestamp, TimestampNs, Timestamptz, F64,
 };
 use risingwave_expr::{function, ExprError, Result};
 
@@ -43,7 +43,7 @@ fn extract_from_datelike(date: impl Datelike, unit: Unit) -> Decimal {
 }
 
 /// Extract field from `Timelike`.
-fn extract_from_timelike(time: impl Timelike, unit: Unit) -> Decimal {
+fn extract_from_timelike_ns(time: impl Timelike, unit: Unit) -> Decimal {
     let nanos = || time.second() as u64 * 1_000_000_000 + time.nanosecond() as u64;
     match unit {
         Hour => time.hour().into(),
@@ -61,6 +61,24 @@ fn extract_from_timelike(time: impl Timelike, unit: Unit) -> Decimal {
     }
 }
 
+/// Extract field from `Timelike`.
+fn extract_from_timelike(time: impl Timelike, unit: Unit) -> Decimal {
+    let usecs = || time.second() as u64 * 1_000_000 + (time.nanosecond() / 1000) as u64;
+    match unit {
+        Hour => time.hour().into(),
+        Minute => time.minute().into(),
+        Second => Decimal::from_i128_with_scale(usecs() as i128, 6),
+        Millisecond => Decimal::from_i128_with_scale(usecs() as i128, 3),
+        Microsecond => usecs().into(),
+        Epoch => {
+            let usecs = time.num_seconds_from_midnight() as u64 * 1_000_000
+                + (time.nanosecond() / 1000) as u64;
+            Decimal::from_i128_with_scale(usecs as i128, 6)
+        }
+        u => unreachable!("invalid unit {:?} for time", u),
+    }
+}
+
 #[function(
     "extract(varchar, date) -> decimal",
     prebuild = "Unit::from_str($0)?.ensure_date()?"
@@ -113,27 +131,27 @@ fn extract_from_timestamp(timestamp: Timestamp, unit: &Unit) -> Decimal {
     "extract(varchar, timestamp_ns) -> decimal",
     prebuild = "Unit::from_str($0)?.ensure_timestamp()?"
 )]
-fn extract_from_timestamp_ns(timestamp_nano: TimestampNanosecond, unit: &Unit) -> Decimal {
+fn extract_from_timestamp_ns(timestamp_ns: TimestampNs, unit: &Unit) -> Decimal {
     match unit {
         Epoch => {
-            if let Some(nanos) = timestamp_nano.0.and_utc().timestamp_nanos_opt() {
+            if let Some(nanos) = timestamp_ns.0.and_utc().timestamp_nanos_opt() {
                 Decimal::from_i128_with_scale(nanos as i128, 9)
             } else {
-                let micros = timestamp_nano.0.and_utc().timestamp_micros();
+                let micros = timestamp_ns.0.and_utc().timestamp_micros();
                 Decimal::from_i128_with_scale(micros as i128, 6)
             }
         }
         Julian => {
-            let epoch = if let Some(nanos) = timestamp_nano.0.and_utc().timestamp_nanos_opt() {
+            let epoch = if let Some(nanos) = timestamp_ns.0.and_utc().timestamp_nanos_opt() {
                 Decimal::from_i128_with_scale(nanos as i128, 9)
             } else {
-                let epoch = timestamp_nano.0.and_utc().timestamp_micros();
+                let epoch = timestamp_ns.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_nano.0.date(), *unit),
-        _ if unit.is_time_unit() => extract_from_timelike(timestamp_nano.0.time(), *unit),
+        _ if unit.is_date_unit() => extract_from_datelike(timestamp_ns.0.date(), *unit),
+        _ if unit.is_time_unit() => extract_from_timelike_ns(timestamp_ns.0.time(), *unit),
         u => unreachable!("invalid unit {:?} for timestamp", u),
     }
 }
@@ -497,9 +515,9 @@ mod tests {
         assert_eq!(extract(Doy), "326");
         assert_eq!(extract(Hour), "12");
         assert_eq!(extract(Minute), "4");
-        assert_eq!(extract(Second), "2.575400000");
-        assert_eq!(extract(Millisecond), "2575.400000");
-        assert_eq!(extract(Microsecond), "2575400.000");
+        assert_eq!(extract(Second), "2.575400");
+        assert_eq!(extract(Millisecond), "2575.400");
+        assert_eq!(extract(Microsecond), "2575400");
         assert_eq!(extract(Epoch), "1637582642.575400");
         assert_eq!(extract(Julian), "2459541.5028075856481481481481");
     }
diff --git a/src/expr/impl/src/scalar/make_time.rs b/src/expr/impl/src/scalar/make_time.rs
index 3a0e959a3a110..5e96cd9b04600 100644
--- a/src/expr/impl/src/scalar/make_time.rs
+++ b/src/expr/impl/src/scalar/make_time.rs
@@ -13,9 +13,7 @@
 // limitations under the License.
 
 use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
-use risingwave_common::types::{
-    Date, FloatExt, Time, Timestamp, TimestampNanosecond, Timestamptz, F64,
-};
+use risingwave_common::types::{Date, FloatExt, Time, Timestamp, TimestampNs, Timestamptz, F64};
 use risingwave_expr::expr_context::TIME_ZONE;
 use risingwave_expr::{capture_context, function, ExprError, Result};
 
@@ -80,7 +78,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_ns(hour, min, sec)?))
+    Ok(Time(make_naive_time_ms(hour, min, sec)?))
 }
 
 // year int, month int, day int, hour int, min int, sec double precision
@@ -107,8 +105,8 @@ pub fn make_timestamp_ns(
     hour: i32,
     min: i32,
     sec: F64,
-) -> Result<TimestampNanosecond> {
-    Ok(TimestampNanosecond(NaiveDateTime::new(
+) -> Result<TimestampNs> {
+    Ok(TimestampNs(NaiveDateTime::new(
         make_naive_date(year, month, day)?,
         make_naive_time_ns(hour, min, sec)?,
     )))
diff --git a/src/expr/impl/src/scalar/timestamptz.rs b/src/expr/impl/src/scalar/timestamptz.rs
index 1dbad92c004da..3771507eb90b6 100644
--- a/src/expr/impl/src/scalar/timestamptz.rs
+++ b/src/expr/impl/src/scalar/timestamptz.rs
@@ -17,8 +17,7 @@ use std::fmt::Write;
 use chrono::LocalResult;
 use num_traits::CheckedNeg;
 use risingwave_common::types::{
-    write_date_time_tz, CheckedAdd, Interval, IntoOrdered, Timestamp, TimestampNanosecond,
-    Timestamptz, F64,
+    write_date_time_tz, CheckedAdd, Interval, IntoOrdered, Timestamp, TimestampNs, Timestamptz, F64,
 };
 use risingwave_expr::{function, ExprError, Result};
 use thiserror_ext::AsReport;
@@ -51,10 +50,7 @@ pub fn timestamptz_at_time_zone(input: Timestamptz, time_zone: &str) -> Result<T
 }
 
 #[function("at_time_zone(timestamp_ns, varchar) -> timestamptz")]
-pub fn timestamp_ns_at_time_zone(
-    input: TimestampNanosecond,
-    time_zone: &str,
-) -> Result<Timestamptz> {
+pub fn timestamp_ns_at_time_zone(input: TimestampNs, time_zone: &str) -> Result<Timestamptz> {
     timestamp_at_time_zone(input.into(), time_zone)
 }
 
diff --git a/src/expr/impl/src/scalar/to_char.rs b/src/expr/impl/src/scalar/to_char.rs
index 5ad078a69ecda..fe32aa0c9ce54 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, TimestampNanosecond, Timestamptz};
+use risingwave_common::types::{Interval, Timestamp, TimestampNs, Timestamptz};
 use risingwave_expr::{function, ExprError, Result};
 
 use super::timestamptz::time_zone_err;
@@ -198,11 +198,7 @@ fn timestamp_to_char(data: Timestamp, pattern: &ChronoPattern, writer: &mut impl
     "to_char(timestamp_ns, varchar) -> varchar",
     prebuild = "ChronoPattern::compile($1)"
 )]
-fn timestamp_ns_to_char(
-    data: TimestampNanosecond,
-    pattern: &ChronoPattern,
-    writer: &mut impl Write,
-) {
+fn timestamp_ns_to_char(data: TimestampNs, pattern: &ChronoPattern, writer: &mut impl Write) {
     let format = data.0.format_with_items(pattern.borrow_dependent().iter());
     write!(writer, "{}", format).unwrap();
 }
diff --git a/src/expr/impl/src/scalar/to_jsonb.rs b/src/expr/impl/src/scalar/to_jsonb.rs
index c934b3407a905..6a70d00c2993f 100644
--- a/src/expr/impl/src/scalar/to_jsonb.rs
+++ b/src/expr/impl/src/scalar/to_jsonb.rs
@@ -17,8 +17,7 @@ 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, TimestampNanosecond, Timestamptz, ToText,
-    F32, F64,
+    ScalarRefImpl, Serial, StructRef, Time, Timestamp, TimestampNs, Timestamptz, ToText, F32, F64,
 };
 use risingwave_common::util::iter_util::ZipEqDebug;
 use risingwave_expr::expr::Context;
@@ -67,7 +66,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),
+            TimestampNs(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),
@@ -196,7 +195,7 @@ impl ToJsonb for Timestamp {
     }
 }
 
-impl ToJsonb for TimestampNanosecond {
+impl ToJsonb for TimestampNs {
     fn add_to(self, _: &DataType, builder: &mut Builder) -> Result<()> {
         builder.display(format_args!("{}T{}", self.0.date(), self.0.time()));
         Ok(())
diff --git a/src/expr/impl/src/udf/wasm.rs b/src/expr/impl/src/udf/wasm.rs
index 0d11491bb6df4..823f135dc8afe 100644
--- a/src/expr/impl/src/udf/wasm.rs
+++ b/src/expr/impl/src/udf/wasm.rs
@@ -264,7 +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::TimestampNs => "timestamp_ns".to_owned(),
         DataType::Timestamptz => "timestamptz".to_owned(),
         DataType::Interval => "interval".to_owned(),
         DataType::Decimal => "decimal".to_owned(),
diff --git a/src/expr/macro/src/lib.rs b/src/expr/macro/src/lib.rs
index b554fcbbd98e4..34898be4b1f0e 100644
--- a/src/expr/macro/src/lib.rs
+++ b/src/expr/macro/src/lib.rs
@@ -386,7 +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        |
+/// | `timestamp_ns`   | `timestamp_ns`        | `TimestampNs`   | `TimestampNs`        | 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 0c8ac019c7c0a..f809938e52d1a 100644
--- a/src/expr/macro/src/types.rs
+++ b/src/expr/macro/src/types.rs
@@ -28,7 +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
+    timestamp_ns   TimestampNs   TimestampNsArray      TimestampNs       TimestampNs           y
     timestamptz Timestamptz TimestamptzArray    Timestamptz     Timestamptz         y
     interval    Interval    IntervalArray       Interval        Interval            y
     varchar     Varchar     Utf8Array           Box<str>        &str                _
@@ -142,7 +142,9 @@ pub fn min_compatible_type(types: &[impl AsRef<str>]) -> &str {
         ("decimal", "decimal") => "decimal",
 
         ("date", "timestamp") => "timestamp",
+        ("date", "timestamp_ns") => "timestamp_ns",
         ("timestamp", "date") => "timestamp",
+        ("timestamp_ns", "date") => "timestamp_ns",
         ("time", "interval") => "interval",
         ("interval", "time") => "interval",
 
diff --git a/src/frontend/src/binder/expr/mod.rs b/src/frontend/src/binder/expr/mod.rs
index 6eef0833ad9e3..05d5e8a32f9d4 100644
--- a/src/frontend/src/binder/expr/mod.rs
+++ b/src/frontend/src/binder/expr/mod.rs
@@ -1064,7 +1064,7 @@ pub fn bind_data_type(data_type: &AstDataType) -> Result<DataType> {
                 "float4" => DataType::Float32,
                 "float8" => DataType::Float64,
                 "timestamptz" => DataType::Timestamptz,
-                "timestamp_ns" => DataType::TimestampNanosecond,
+                "timestamp_ns" => DataType::TimestampNs,
                 "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 e5a6701a96c5f..03ff17fe58361 100644
--- a/src/frontend/src/expr/literal.rs
+++ b/src/frontend/src/expr/literal.rs
@@ -49,7 +49,7 @@ impl std::fmt::Debug for Literal {
                     | DataType::Bytea
                     | DataType::Date
                     | DataType::Timestamp
-                    | DataType::TimestampNanosecond
+                    | DataType::TimestampNs
                     | DataType::Timestamptz
                     | DataType::Time
                     | DataType::Interval
diff --git a/src/frontend/src/expr/session_timezone.rs b/src/frontend/src/expr/session_timezone.rs
index 06b6cb7cf7fff..c8d0f1ac89abd 100644
--- a/src/frontend/src/expr/session_timezone.rs
+++ b/src/frontend/src/expr/session_timezone.rs
@@ -107,14 +107,14 @@ impl SessionTimezone {
                     }
                     (DataType::Date, DataType::Timestamptz)
                     | (DataType::Timestamp, DataType::Timestamptz)
-                    | (DataType::TimestampNanosecond, DataType::Timestamptz) => {
+                    | (DataType::TimestampNs, 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::TimestampNanosecond) => {
+                    | (DataType::Timestamptz, DataType::TimestampNs) => {
                         input = self.at_timezone(input);
                         input = input.cast_explicit(return_type).unwrap();
                         Some(input)
@@ -144,7 +144,7 @@ impl SessionTimezone {
                     if matches!(inputs[(idx + 1) % 2].return_type(), DataType::Timestamptz)
                         && matches!(
                             inputs[idx % 2].return_type(),
-                            DataType::Date | DataType::Timestamp | DataType::TimestampNanosecond
+                            DataType::Date | DataType::Timestamp
                         )
                     {
                         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 9a5fb5ecdc897..977a2172903cd 100644
--- a/src/frontend/src/expr/type_inference/cast.rs
+++ b/src/frontend/src/expr/type_inference/cast.rs
@@ -306,24 +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 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
+        (". 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 .", TimestampNs), // H
     ];
     let mut map = BTreeMap::new();
     for (row, source) in CAST_TABLE {
diff --git a/src/frontend/src/optimizer/property/monotonicity.rs b/src/frontend/src/optimizer/property/monotonicity.rs
index e610e5472df3a..db020cdb2c64c 100644
--- a/src/frontend/src/optimizer/property/monotonicity.rs
+++ b/src/frontend/src/optimizer/property/monotonicity.rs
@@ -232,8 +232,7 @@ 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
-                        || func_call.inputs()[0].return_type() == DataType::TimestampNanosecond)
+                    if (func_call.inputs()[0].return_type() == DataType::Timestamp
                         && 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 48fecd4d11bc8..1faecbed755e6 100644
--- a/src/frontend/src/optimizer/rule/index_selection_rule.rs
+++ b/src/frontend/src/optimizer/rule/index_selection_rule.rs
@@ -54,8 +54,7 @@ use std::rc::Rc;
 use itertools::Itertools;
 use risingwave_common::catalog::Schema;
 use risingwave_common::types::{
-    DataType, Date, Decimal, Int256, Interval, Serial, Time, Timestamp, TimestampNanosecond,
-    Timestamptz,
+    DataType, Date, Decimal, Int256, Interval, Serial, Time, Timestamp, TimestampNs, Timestamptz,
 };
 use risingwave_common::util::iter_util::ZipEqFast;
 use risingwave_pb::plan_common::JoinType;
@@ -764,7 +763,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::TimestampNs => size_of::<TimestampNs>(),
             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 e27bd74b67396..189130c2b4199 100644
--- a/src/frontend/src/utils/data_type.rs
+++ b/src/frontend/src/utils/data_type.rs
@@ -32,9 +32,7 @@ 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::TimestampNs => 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 882823d269fb2..999f659927d87 100644
--- a/src/tests/sqlsmith/src/sql_gen/types.rs
+++ b/src/tests/sqlsmith/src/sql_gen/types.rs
@@ -39,7 +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::TimestampNs => AstDataType::Custom(vec!["timestamp_ns".into()].into()),
         DataType::Timestamptz => AstDataType::Timestamp(true),
         DataType::Time => AstDataType::Time(false),
         DataType::Interval => AstDataType::Interval,