fix(expr): align timestamp(tz)'s output format with pg's

e2e_test/batch/basic/make_time.slt.part

@@ -9,15 +9,20 @@ SELECT make_timestamptz(1973, 07, 15, 08, 15, 55.33);
 query T
 SELECT make_timestamptz(-1973, 07, 15, 08, 15, 55.33);
 ----
--1972-07-15 08:15:55.330+00:00
+1973-07-15 08:15:55.330+00:00 BC
+
+query T
+SELECT make_timestamptz(20240, 1, 26, 14, 20, 26);
+----
+20240-01-26 14:20:26+00:00
 
 query error Invalid parameter year, month, day: invalid date: -3-2-29
 SELECT make_timestamptz(-4, 02, 29, 08, 15, 55.33);
 
 query T
 SELECT make_timestamptz(-5, 02, 29, 08, 15, 55.33);
 ----
--0004-02-29 08:15:55.330+00:00
+0005-02-29 08:15:55.330+00:00 BC
 
 query error Invalid parameter sec: invalid sec: -55.33
 SELECT make_timestamptz(1973, 07, 15, 08, 15, -55.33);
@@ -105,6 +110,11 @@ SELECT make_date(2024, 1, 26);
 ----
 2024-01-26
 
+query T
+SELECT make_date(20240, 1, 26);
+----
+20240-01-26
+
 query T
 SELECT make_date(-2024, 1, 26);
 ----
@@ -146,15 +156,30 @@ SELECT make_timestamp(2024, 1, 26, 14, 20, 26);
 ----
 2024-01-26 14:20:26
 
+query T
+SELECT make_timestamp(20240, 1, 26, 14, 20, 26);
+----
+20240-01-26 14:20:26
+
 query T
 SELECT make_timestamp(-1973, 07, 15, 08, 15, 55.33);
 ----
--1972-07-15 08:15:55.330
+1973-07-15 08:15:55.330 BC
 
 query error Invalid parameter year, month, day: invalid date: -3-2-29
 SELECT make_timestamp(-4, 02, 29, 08, 15, 55.33);
 
 query T
 SELECT make_timestamp(-5, 02, 29, 08, 15, 55.33);
 ----
--0004-02-29 08:15:55.330
+0005-02-29 08:15:55.330 BC
+
+query T
+select '0001-01-01 12:34:56'::timestamp - '10 year'::interval;
+----
+0010-01-01 12:34:56 BC
+
+query T
+select '0001-01-01 12:34:56'::timestamptz - '10 year'::interval;
+----
+0010-01-01 12:34:56+00:00 BC

src/common/src/types/datetime.rs

@@ -328,17 +328,15 @@ impl ToText for Date {
     /// ```
     fn write<W: std::fmt::Write>(&self, f: &mut W) -> std::fmt::Result {
         let (ce, year) = self.0.year_ce();
-        if ce {
-            write!(f, "{}", self.0)
-        } else {
-            write!(
-                f,
-                "{:04}-{:02}-{:02} BC",
-                year,
-                self.0.month(),
-                self.0.day()
-            )
-        }
+        let suffix = if ce { "" } else { " BC" };
+        write!(
+            f,
+            "{:04}-{:02}-{:02}{}",
+            year,
+            self.0.month(),
+            self.0.day(),
+            suffix
+        )
     }
 
     fn write_with_type<W: std::fmt::Write>(&self, ty: &DataType, f: &mut W) -> std::fmt::Result {
@@ -364,7 +362,17 @@ impl ToText for Time {
 
 impl ToText for Timestamp {
     fn write<W: std::fmt::Write>(&self, f: &mut W) -> std::fmt::Result {
-        write!(f, "{}", self.0)
+        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 {

src/common/src/types/timestamptz.rs

@@ -17,7 +17,7 @@ use std::io::Write;
 use std::str::FromStr;
 
 use bytes::{Bytes, BytesMut};
-use chrono::{TimeZone, Utc};
+use chrono::{DateTime, Datelike, TimeZone, Utc};
 use chrono_tz::Tz;
 use postgres_types::{accepts, to_sql_checked, IsNull, ToSql, Type};
 use serde::{Deserialize, Serialize};
@@ -201,6 +201,26 @@ impl std::fmt::Display for Timestamptz {
     }
 }
 
+pub fn write_date_time_tz(
+    instant_local: DateTime<Tz>,
+    writer: &mut impl std::fmt::Write,
+) -> std::fmt::Result {
+    let date = instant_local.date_naive();
+    let (ce, year) = date.year_ce();
+    write!(
+        writer,
+        "{:04}-{:02}-{:02} {}",
+        year,
+        date.month(),
+        date.day(),
+        instant_local.format(if ce {
+            "%H:%M:%S%.f%:z"
+        } else {
+            "%H:%M:%S%.f%:z BC"
+        })
+    )
+}
+
 #[cfg(test)]
 mod test {
     use super::*;

src/expr/impl/Cargo.toml

@@ -25,6 +25,7 @@ chrono = { version = "0.4", default-features = false, features = [
     "clock",
     "std",
 ] }
+chrono-tz = { version = "0.8", features = ["case-insensitive"] }
 fancy-regex = "0.13"
 futures-async-stream = { workspace = true }
 futures-util = "0.3"

src/expr/impl/src/scalar/make_time.rs

@@ -129,8 +129,6 @@ mod tests {
     use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
     use risingwave_common::types::{Date, Timestamp};
 
-    /// This test is to testify that our `Date` expressess a year `-X` as `X+1 BC`, while `Timestamp` expresses it as `-X`.
-    /// Can be removed if we change the `ToText` implementation of `Date` or `Timestamp`.
     #[test]
     fn test_naive_date_and_time() {
         let year = -1973;
@@ -154,7 +152,7 @@ mod tests {
         let date_time = Timestamp(NaiveDateTime::new(naive_date, naive_time));
         assert_eq!(
             date_time.to_string(),
-            String::from("-1973-02-02 12:34:56.789")
+            String::from("1974-02-02 12:34:56.789 BC")
         );
     }
 }

src/expr/impl/src/scalar/timestamptz.rs

@@ -15,7 +15,9 @@
 use std::fmt::Write;
 
 use num_traits::CheckedNeg;
-use risingwave_common::types::{CheckedAdd, Interval, IntoOrdered, Timestamp, Timestamptz, F64};
+use risingwave_common::types::{
+    write_date_time_tz, CheckedAdd, Interval, IntoOrdered, Timestamp, Timestamptz, F64,
+};
 use risingwave_expr::{function, ExprError, Result};
 use thiserror_ext::AsReport;
 
@@ -72,13 +74,7 @@ pub fn timestamptz_to_string(
 ) -> Result<()> {
     let time_zone = Timestamptz::lookup_time_zone(time_zone).map_err(time_zone_err)?;
     let instant_local = elem.to_datetime_in_zone(time_zone);
-    write!(
-        writer,
-        "{}",
-        instant_local.format("%Y-%m-%d %H:%M:%S%.f%:z")
-    )
-    .map_err(|e| ExprError::Internal(e.into()))?;
-    Ok(())
+    write_date_time_tz(instant_local, writer).map_err(|e| ExprError::Internal(e.into()))
 }
 
 // Tries to interpret the string with a timezone, and if failing, tries to interpret the string as a

src/frontend/src/handler/util.rs

@@ -18,7 +18,7 @@ use std::sync::Arc;
 use std::task::{Context, Poll};
 
 use anyhow::Context as _;
-use bytes::Bytes;
+use bytes::{Bytes, BytesMut};
 use futures::Stream;
 use itertools::Itertools;
 use pgwire::pg_field_descriptor::PgFieldDescriptor;
@@ -29,7 +29,7 @@ use pin_project_lite::pin_project;
 use risingwave_common::array::DataChunk;
 use risingwave_common::catalog::Field;
 use risingwave_common::row::Row as _;
-use risingwave_common::types::{DataType, ScalarRefImpl, Timestamptz};
+use risingwave_common::types::{write_date_time_tz, DataType, ScalarRefImpl, Timestamptz};
 use risingwave_common::util::iter_util::ZipEqFast;
 use risingwave_connector::source::iceberg::ICEBERG_CONNECTOR;
 use risingwave_connector::source::KAFKA_CONNECTOR;
@@ -139,10 +139,9 @@ fn timestamptz_to_string_with_session_data(
     let tz = d.into_timestamptz();
     let time_zone = Timestamptz::lookup_time_zone(&session_data.timezone).unwrap();
     let instant_local = tz.to_datetime_in_zone(time_zone);
-    instant_local
-        .format("%Y-%m-%d %H:%M:%S%.f%:z")
-        .to_string()
-        .into()
+    let mut result_string = BytesMut::new();
+    write_date_time_tz(instant_local, &mut result_string).unwrap();
+    result_string.into()
 }
 
 fn to_pg_rows(