diff --git a/src/common/function/src/function.rs b/src/common/function/src/function.rs
index 3225183bd4..3b6a51b028 100644
--- a/src/common/function/src/function.rs
+++ b/src/common/function/src/function.rs
@@ -15,21 +15,22 @@
 use std::fmt;
 use std::sync::Arc;
 
-use chrono_tz::Tz;
 use common_query::error::Result;
 use common_query::prelude::Signature;
+use common_time::timezone::get_timezone;
+use common_time::Timezone;
 use datatypes::data_type::ConcreteDataType;
 use datatypes::vectors::VectorRef;
 
 #[derive(Clone)]
 pub struct FunctionContext {
-    pub tz: Tz,
+    pub timezone: Timezone,
 }
 
 impl Default for FunctionContext {
     fn default() -> Self {
         Self {
-            tz: "UTC".parse::<Tz>().unwrap(),
+            timezone: get_timezone(None).clone(),
         }
     }
 }
diff --git a/src/common/function/src/scalars/date.rs b/src/common/function/src/scalars/date.rs
index 86b0c7db62..4b8e714ec5 100644
--- a/src/common/function/src/scalars/date.rs
+++ b/src/common/function/src/scalars/date.rs
@@ -14,9 +14,11 @@
 
 use std::sync::Arc;
 mod date_add;
+mod date_format;
 mod date_sub;
 
 use date_add::DateAddFunction;
+use date_format::DateFormatFunction;
 use date_sub::DateSubFunction;
 
 use crate::function_registry::FunctionRegistry;
@@ -27,5 +29,6 @@ impl DateFunction {
     pub fn register(registry: &FunctionRegistry) {
         registry.register(Arc::new(DateAddFunction));
         registry.register(Arc::new(DateSubFunction));
+        registry.register(Arc::new(DateFormatFunction));
     }
 }
diff --git a/src/common/function/src/scalars/date/date_format.rs b/src/common/function/src/scalars/date/date_format.rs
new file mode 100644
index 0000000000..d94f115e54
--- /dev/null
+++ b/src/common/function/src/scalars/date/date_format.rs
@@ -0,0 +1,306 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::fmt;
+
+use common_error::ext::BoxedError;
+use common_query::error::{self, InvalidFuncArgsSnafu, Result, UnsupportedInputDataTypeSnafu};
+use common_query::prelude::Signature;
+use datatypes::prelude::{ConcreteDataType, MutableVector, ScalarVectorBuilder};
+use datatypes::vectors::{StringVectorBuilder, VectorRef};
+use snafu::{ensure, ResultExt};
+
+use crate::function::{Function, FunctionContext};
+use crate::helper;
+
+/// A function that formats timestamp/date/datetime into string by the format
+#[derive(Clone, Debug, Default)]
+pub struct DateFormatFunction;
+
+const NAME: &str = "date_format";
+
+impl Function for DateFormatFunction {
+    fn name(&self) -> &str {
+        NAME
+    }
+
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::string_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        helper::one_of_sigs2(
+            vec![
+                ConcreteDataType::date_datatype(),
+                ConcreteDataType::datetime_datatype(),
+                ConcreteDataType::timestamp_second_datatype(),
+                ConcreteDataType::timestamp_millisecond_datatype(),
+                ConcreteDataType::timestamp_microsecond_datatype(),
+                ConcreteDataType::timestamp_nanosecond_datatype(),
+            ],
+            vec![ConcreteDataType::string_datatype()],
+        )
+    }
+
+    fn eval(&self, func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure!(
+            columns.len() == 2,
+            InvalidFuncArgsSnafu {
+                err_msg: format!(
+                    "The length of the args is not correct, expect 2, have: {}",
+                    columns.len()
+                ),
+            }
+        );
+
+        let left = &columns[0];
+        let formats = &columns[1];
+
+        let size = left.len();
+        let left_datatype = columns[0].data_type();
+        let mut results = StringVectorBuilder::with_capacity(size);
+
+        match left_datatype {
+            ConcreteDataType::Timestamp(_) => {
+                for i in 0..size {
+                    let ts = left.get(i).as_timestamp();
+                    let format = formats.get(i).as_string();
+
+                    let result = match (ts, format) {
+                        (Some(ts), Some(fmt)) => Some(
+                            ts.as_formatted_string(&fmt, Some(&func_ctx.timezone))
+                                .map_err(BoxedError::new)
+                                .context(error::ExecuteSnafu)?,
+                        ),
+                        _ => None,
+                    };
+
+                    results.push(result.as_deref());
+                }
+            }
+            ConcreteDataType::Date(_) => {
+                for i in 0..size {
+                    let date = left.get(i).as_date();
+                    let format = formats.get(i).as_string();
+
+                    let result = match (date, format) {
+                        (Some(date), Some(fmt)) => date
+                            .as_formatted_string(&fmt, Some(&func_ctx.timezone))
+                            .map_err(BoxedError::new)
+                            .context(error::ExecuteSnafu)?,
+                        _ => None,
+                    };
+
+                    results.push(result.as_deref());
+                }
+            }
+            ConcreteDataType::DateTime(_) => {
+                for i in 0..size {
+                    let datetime = left.get(i).as_datetime();
+                    let format = formats.get(i).as_string();
+
+                    let result = match (datetime, format) {
+                        (Some(datetime), Some(fmt)) => datetime
+                            .as_formatted_string(&fmt, Some(&func_ctx.timezone))
+                            .map_err(BoxedError::new)
+                            .context(error::ExecuteSnafu)?,
+                        _ => None,
+                    };
+
+                    results.push(result.as_deref());
+                }
+            }
+            _ => {
+                return UnsupportedInputDataTypeSnafu {
+                    function: NAME,
+                    datatypes: columns.iter().map(|c| c.data_type()).collect::<Vec<_>>(),
+                }
+                .fail();
+            }
+        }
+
+        Ok(results.to_vector())
+    }
+}
+
+impl fmt::Display for DateFormatFunction {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "DATE_FORMAT")
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use common_query::prelude::{TypeSignature, Volatility};
+    use datatypes::prelude::{ConcreteDataType, ScalarVector};
+    use datatypes::value::Value;
+    use datatypes::vectors::{DateTimeVector, DateVector, StringVector, TimestampSecondVector};
+
+    use super::{DateFormatFunction, *};
+
+    #[test]
+    fn test_date_format_misc() {
+        let f = DateFormatFunction;
+        assert_eq!("date_format", f.name());
+        assert_eq!(
+            ConcreteDataType::string_datatype(),
+            f.return_type(&[ConcreteDataType::timestamp_microsecond_datatype()])
+                .unwrap()
+        );
+        assert_eq!(
+            ConcreteDataType::string_datatype(),
+            f.return_type(&[ConcreteDataType::timestamp_second_datatype()])
+                .unwrap()
+        );
+        assert_eq!(
+            ConcreteDataType::string_datatype(),
+            f.return_type(&[ConcreteDataType::date_datatype()]).unwrap()
+        );
+        assert_eq!(
+            ConcreteDataType::string_datatype(),
+            f.return_type(&[ConcreteDataType::datetime_datatype()])
+                .unwrap()
+        );
+        assert!(matches!(f.signature(),
+                         Signature {
+                             type_signature: TypeSignature::OneOf(sigs),
+                             volatility: Volatility::Immutable
+                         } if  sigs.len() == 6));
+    }
+
+    #[test]
+    fn test_timestamp_date_format() {
+        let f = DateFormatFunction;
+
+        let times = vec![Some(123), None, Some(42), None];
+        let formats = vec![
+            "%Y-%m-%d %T.%3f",
+            "%Y-%m-%d %T.%3f",
+            "%Y-%m-%d %T.%3f",
+            "%Y-%m-%d %T.%3f",
+        ];
+        let results = [
+            Some("1970-01-01 00:02:03.000"),
+            None,
+            Some("1970-01-01 00:00:42.000"),
+            None,
+        ];
+
+        let time_vector = TimestampSecondVector::from(times.clone());
+        let interval_vector = StringVector::from_vec(formats);
+        let args: Vec<VectorRef> = vec![Arc::new(time_vector), Arc::new(interval_vector)];
+        let vector = f.eval(FunctionContext::default(), &args).unwrap();
+
+        assert_eq!(4, vector.len());
+        for (i, _t) in times.iter().enumerate() {
+            let v = vector.get(i);
+            let result = results.get(i).unwrap();
+
+            if result.is_none() {
+                assert_eq!(Value::Null, v);
+                continue;
+            }
+            match v {
+                Value::String(s) => {
+                    assert_eq!(s.as_utf8(), result.unwrap());
+                }
+                _ => unreachable!(),
+            }
+        }
+    }
+
+    #[test]
+    fn test_date_date_format() {
+        let f = DateFormatFunction;
+
+        let dates = vec![Some(123), None, Some(42), None];
+        let formats = vec![
+            "%Y-%m-%d %T.%3f",
+            "%Y-%m-%d %T.%3f",
+            "%Y-%m-%d %T.%3f",
+            "%Y-%m-%d %T.%3f",
+        ];
+        let results = [
+            Some("1970-05-04 00:00:00.000"),
+            None,
+            Some("1970-02-12 00:00:00.000"),
+            None,
+        ];
+
+        let date_vector = DateVector::from(dates.clone());
+        let interval_vector = StringVector::from_vec(formats);
+        let args: Vec<VectorRef> = vec![Arc::new(date_vector), Arc::new(interval_vector)];
+        let vector = f.eval(FunctionContext::default(), &args).unwrap();
+
+        assert_eq!(4, vector.len());
+        for (i, _t) in dates.iter().enumerate() {
+            let v = vector.get(i);
+            let result = results.get(i).unwrap();
+
+            if result.is_none() {
+                assert_eq!(Value::Null, v);
+                continue;
+            }
+            match v {
+                Value::String(s) => {
+                    assert_eq!(s.as_utf8(), result.unwrap());
+                }
+                _ => unreachable!(),
+            }
+        }
+    }
+
+    #[test]
+    fn test_datetime_date_format() {
+        let f = DateFormatFunction;
+
+        let dates = vec![Some(123), None, Some(42), None];
+        let formats = vec![
+            "%Y-%m-%d %T.%3f",
+            "%Y-%m-%d %T.%3f",
+            "%Y-%m-%d %T.%3f",
+            "%Y-%m-%d %T.%3f",
+        ];
+        let results = [
+            Some("1970-01-01 00:00:00.123"),
+            None,
+            Some("1970-01-01 00:00:00.042"),
+            None,
+        ];
+
+        let date_vector = DateTimeVector::from(dates.clone());
+        let interval_vector = StringVector::from_vec(formats);
+        let args: Vec<VectorRef> = vec![Arc::new(date_vector), Arc::new(interval_vector)];
+        let vector = f.eval(FunctionContext::default(), &args).unwrap();
+
+        assert_eq!(4, vector.len());
+        for (i, _t) in dates.iter().enumerate() {
+            let v = vector.get(i);
+            let result = results.get(i).unwrap();
+
+            if result.is_none() {
+                assert_eq!(Value::Null, v);
+                continue;
+            }
+            match v {
+                Value::String(s) => {
+                    assert_eq!(s.as_utf8(), result.unwrap());
+                }
+                _ => unreachable!(),
+            }
+        }
+    }
+}
diff --git a/src/common/function/src/scalars/expression/ctx.rs b/src/common/function/src/scalars/expression/ctx.rs
index 65844548a2..362997ab24 100644
--- a/src/common/function/src/scalars/expression/ctx.rs
+++ b/src/common/function/src/scalars/expression/ctx.rs
@@ -12,20 +12,20 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use chrono_tz::Tz;
 use common_query::error::Error;
+use common_time::timezone::get_timezone;
+use common_time::Timezone;
 
 pub struct EvalContext {
-    _tz: Tz,
+    pub timezone: Timezone,
     pub error: Option<Error>,
 }
 
 impl Default for EvalContext {
     fn default() -> Self {
-        let tz = "UTC".parse::<Tz>().unwrap();
         Self {
             error: None,
-            _tz: tz,
+            timezone: get_timezone(None).clone(),
         }
     }
 }
diff --git a/src/common/function/src/scalars/udf.rs b/src/common/function/src/scalars/udf.rs
index da67f321bf..5b91ad1302 100644
--- a/src/common/function/src/scalars/udf.rs
+++ b/src/common/function/src/scalars/udf.rs
@@ -34,6 +34,8 @@ pub fn create_udf(func: FunctionRef) -> ScalarUdf {
 
     let func_cloned = func.clone();
     let fun: ScalarFunctionImplementation = Arc::new(move |args: &[ColumnarValue]| {
+        // FIXME(dennis): set the timezone into function context
+        // Question: how to get the timezone from the query context?
         let func_ctx = FunctionContext::default();
 
         let len = args
diff --git a/src/common/query/src/error.rs b/src/common/query/src/error.rs
index 25f9d3c715..86a3d5e958 100644
--- a/src/common/query/src/error.rs
+++ b/src/common/query/src/error.rs
@@ -163,6 +163,12 @@ pub enum Error {
         source: DataTypeError,
     },
 
+    #[snafu(display("Failed to execute function: {source}"))]
+    Execute {
+        location: Location,
+        source: BoxedError,
+    },
+
     #[snafu(display("Invalid function args: {}", err_msg))]
     InvalidFuncArgs { err_msg: String, location: Location },
 }
@@ -201,6 +207,7 @@ impl ErrorExt for Error {
 
             Error::ConvertDfRecordBatchStream { source, .. } => source.status_code(),
             Error::ExecutePhysicalPlan { source, .. } => source.status_code(),
+            Error::Execute { source, .. } => source.status_code(),
         }
     }
 
diff --git a/src/common/time/src/date.rs b/src/common/time/src/date.rs
index 7ff1914543..d4182b7c1b 100644
--- a/src/common/time/src/date.rs
+++ b/src/common/time/src/date.rs
@@ -12,16 +12,18 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::fmt::{Display, Formatter};
+use std::fmt::{Display, Formatter, Write};
 use std::str::FromStr;
 
-use chrono::{Datelike, Days, Months, NaiveDate};
+use chrono::{Datelike, Days, Months, NaiveDate, NaiveTime, TimeZone};
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use snafu::ResultExt;
 
 use crate::error::{Error, ParseDateStrSnafu, Result};
 use crate::interval::Interval;
+use crate::timezone::get_timezone;
+use crate::Timezone;
 
 const UNIX_EPOCH_FROM_CE: i32 = 719_163;
 
@@ -84,6 +86,40 @@ impl Date {
         NaiveDate::from_num_days_from_ce_opt(UNIX_EPOCH_FROM_CE + self.0)
     }
 
+    /// Format Date for given format and timezone.
+    /// If `tz==None`, the server default timezone will used.
+    pub fn as_formatted_string(
+        self,
+        pattern: &str,
+        timezone: Option<&Timezone>,
+    ) -> Result<Option<String>> {
+        if let Some(v) = self.to_chrono_date() {
+            // Safety: always success
+            let time = NaiveTime::from_hms_nano_opt(0, 0, 0, 0).unwrap();
+            let v = v.and_time(time);
+            let mut formatted = String::new();
+
+            match get_timezone(timezone) {
+                Timezone::Offset(offset) => {
+                    write!(
+                        formatted,
+                        "{}",
+                        offset.from_utc_datetime(&v).format(pattern)
+                    )
+                    .context(crate::error::FormatSnafu { pattern })?;
+                }
+                Timezone::Named(tz) => {
+                    write!(formatted, "{}", tz.from_utc_datetime(&v).format(pattern))
+                        .context(crate::error::FormatSnafu { pattern })?;
+                }
+            }
+
+            return Ok(Some(formatted));
+        }
+
+        Ok(None)
+    }
+
     pub fn to_secs(&self) -> i64 {
         (self.0 as i64) * 24 * 3600
     }
@@ -170,6 +206,37 @@ mod tests {
         assert_eq!(date, Date::from_str(&date.to_string()).unwrap());
     }
 
+    #[test]
+    fn test_as_formatted_string() {
+        let d: Date = 42.into();
+
+        assert_eq!(
+            "1970-02-12",
+            d.as_formatted_string("%Y-%m-%d", None).unwrap().unwrap()
+        );
+        assert_eq!(
+            "1970-02-12 00:00:00",
+            d.as_formatted_string("%Y-%m-%d %H:%M:%S", None)
+                .unwrap()
+                .unwrap()
+        );
+        assert_eq!(
+            "1970-02-12T00:00:00:000",
+            d.as_formatted_string("%Y-%m-%dT%H:%M:%S:%3f", None)
+                .unwrap()
+                .unwrap()
+        );
+        assert_eq!(
+            "1970-02-12T08:00:00:000",
+            d.as_formatted_string(
+                "%Y-%m-%dT%H:%M:%S:%3f",
+                Some(&Timezone::from_tz_string("Asia/Shanghai").unwrap())
+            )
+            .unwrap()
+            .unwrap()
+        );
+    }
+
     #[test]
     pub fn test_from() {
         let d: Date = 42.into();
diff --git a/src/common/time/src/datetime.rs b/src/common/time/src/datetime.rs
index 7dc872b8f2..722467cfab 100644
--- a/src/common/time/src/datetime.rs
+++ b/src/common/time/src/datetime.rs
@@ -12,15 +12,16 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::fmt::{Display, Formatter};
+use std::fmt::{Display, Formatter, Write};
 use std::str::FromStr;
 use std::time::Duration;
 
 use chrono::{Days, LocalResult, Months, NaiveDateTime, TimeZone as ChronoTimeZone, Utc};
 use serde::{Deserialize, Serialize};
+use snafu::ResultExt;
 
 use crate::error::{Error, InvalidDateStrSnafu, Result};
-use crate::timezone::Timezone;
+use crate::timezone::{get_timezone, Timezone};
 use crate::util::{format_utc_datetime, local_datetime_to_utc};
 use crate::{Date, Interval};
 
@@ -110,7 +111,38 @@ impl DateTime {
         NaiveDateTime::from_timestamp_millis(self.0)
     }
 
-    pub fn to_chrono_datetime_with_timezone(&self, tz: Option<Timezone>) -> Option<NaiveDateTime> {
+    /// Format DateTime for given format and timezone.
+    /// If `tz==None`, the server default timezone will used.
+    pub fn as_formatted_string(
+        self,
+        pattern: &str,
+        timezone: Option<&Timezone>,
+    ) -> Result<Option<String>> {
+        if let Some(v) = self.to_chrono_datetime() {
+            let mut formatted = String::new();
+
+            match get_timezone(timezone) {
+                Timezone::Offset(offset) => {
+                    write!(
+                        formatted,
+                        "{}",
+                        offset.from_utc_datetime(&v).format(pattern)
+                    )
+                    .context(crate::error::FormatSnafu { pattern })?;
+                }
+                Timezone::Named(tz) => {
+                    write!(formatted, "{}", tz.from_utc_datetime(&v).format(pattern))
+                        .context(crate::error::FormatSnafu { pattern })?;
+                }
+            }
+
+            return Ok(Some(formatted));
+        }
+
+        Ok(None)
+    }
+
+    pub fn to_chrono_datetime_with_timezone(&self, tz: Option<&Timezone>) -> Option<NaiveDateTime> {
         let datetime = self.to_chrono_datetime();
         datetime.map(|v| match tz {
             Some(Timezone::Offset(offset)) => offset.from_utc_datetime(&v).naive_local(),
@@ -211,6 +243,38 @@ mod tests {
         assert_eq!(28800000, ts);
     }
 
+    #[test]
+    fn test_as_formatted_string() {
+        let d: DateTime = DateTime::new(1000);
+
+        assert_eq!(
+            "1970-01-01",
+            d.as_formatted_string("%Y-%m-%d", None).unwrap().unwrap()
+        );
+        assert_eq!(
+            "1970-01-01 00:00:01",
+            d.as_formatted_string("%Y-%m-%d %H:%M:%S", None)
+                .unwrap()
+                .unwrap()
+        );
+        assert_eq!(
+            "1970-01-01T00:00:01:000",
+            d.as_formatted_string("%Y-%m-%dT%H:%M:%S:%3f", None)
+                .unwrap()
+                .unwrap()
+        );
+
+        assert_eq!(
+            "1970-01-01T08:00:01:000",
+            d.as_formatted_string(
+                "%Y-%m-%dT%H:%M:%S:%3f",
+                Some(&Timezone::from_tz_string("Asia/Shanghai").unwrap())
+            )
+            .unwrap()
+            .unwrap()
+        );
+    }
+
     #[test]
     fn test_from_max_date() {
         let date = Date::new(i32::MAX);
diff --git a/src/common/time/src/error.rs b/src/common/time/src/error.rs
index a1d2256105..c4c025dc92 100644
--- a/src/common/time/src/error.rs
+++ b/src/common/time/src/error.rs
@@ -68,6 +68,14 @@ pub enum Error {
 
     #[snafu(display("Invalid timezone string {raw}"))]
     ParseTimezoneName { raw: String, location: Location },
+
+    #[snafu(display("Failed to format, pattern: {}", pattern))]
+    Format {
+        pattern: String,
+        #[snafu(source)]
+        error: std::fmt::Error,
+        location: Location,
+    },
 }
 
 impl ErrorExt for Error {
@@ -76,6 +84,7 @@ impl ErrorExt for Error {
             Error::ParseDateStr { .. }
             | Error::ParseTimestamp { .. }
             | Error::InvalidTimezoneOffset { .. }
+            | Error::Format { .. }
             | Error::ParseOffsetStr { .. }
             | Error::ParseTimezoneName { .. } => StatusCode::InvalidArguments,
             Error::TimestampOverflow { .. } => StatusCode::Internal,
@@ -93,6 +102,7 @@ impl ErrorExt for Error {
     fn location_opt(&self) -> Option<common_error::snafu::Location> {
         match self {
             Error::ParseTimestamp { location, .. }
+            | Error::Format { location, .. }
             | Error::TimestampOverflow { location, .. }
             | Error::ArithmeticOverflow { location, .. } => Some(*location),
             Error::ParseDateStr { .. }
diff --git a/src/common/time/src/time.rs b/src/common/time/src/time.rs
index fdcc9ee32e..8490195ff6 100644
--- a/src/common/time/src/time.rs
+++ b/src/common/time/src/time.rs
@@ -115,11 +115,11 @@ impl Time {
 
     /// Format Time for given timezone.
     /// When timezone is None, using system timezone by default.
-    pub fn to_timezone_aware_string(&self, tz: Option<Timezone>) -> String {
+    pub fn to_timezone_aware_string(&self, tz: Option<&Timezone>) -> String {
         self.as_formatted_string("%H:%M:%S%.f", tz)
     }
 
-    fn as_formatted_string(self, pattern: &str, timezone: Option<Timezone>) -> String {
+    fn as_formatted_string(self, pattern: &str, timezone: Option<&Timezone>) -> String {
         if let Some(time) = self.to_chrono_time() {
             let date = Utc::now().date_naive();
             let datetime = NaiveDateTime::new(date, time);
@@ -380,37 +380,39 @@ mod tests {
         assert_eq!(
             "08:00:00.001",
             Time::new(1, TimeUnit::Millisecond)
-                .to_timezone_aware_string(Some(Timezone::from_tz_string("SYSTEM").unwrap()))
+                .to_timezone_aware_string(Some(&Timezone::from_tz_string("SYSTEM").unwrap()))
         );
         assert_eq!(
             "08:00:00.001",
             Time::new(1, TimeUnit::Millisecond)
-                .to_timezone_aware_string(Some(Timezone::from_tz_string("+08:00").unwrap()))
+                .to_timezone_aware_string(Some(&Timezone::from_tz_string("+08:00").unwrap()))
         );
         assert_eq!(
             "07:00:00.001",
             Time::new(1, TimeUnit::Millisecond)
-                .to_timezone_aware_string(Some(Timezone::from_tz_string("+07:00").unwrap()))
+                .to_timezone_aware_string(Some(&Timezone::from_tz_string("+07:00").unwrap()))
         );
         assert_eq!(
             "23:00:00.001",
             Time::new(1, TimeUnit::Millisecond)
-                .to_timezone_aware_string(Some(Timezone::from_tz_string("-01:00").unwrap()))
+                .to_timezone_aware_string(Some(&Timezone::from_tz_string("-01:00").unwrap()))
         );
         assert_eq!(
             "08:00:00.001",
-            Time::new(1, TimeUnit::Millisecond)
-                .to_timezone_aware_string(Some(Timezone::from_tz_string("Asia/Shanghai").unwrap()))
+            Time::new(1, TimeUnit::Millisecond).to_timezone_aware_string(Some(
+                &Timezone::from_tz_string("Asia/Shanghai").unwrap()
+            ))
         );
         assert_eq!(
             "00:00:00.001",
             Time::new(1, TimeUnit::Millisecond)
-                .to_timezone_aware_string(Some(Timezone::from_tz_string("UTC").unwrap()))
+                .to_timezone_aware_string(Some(&Timezone::from_tz_string("UTC").unwrap()))
         );
         assert_eq!(
             "03:00:00.001",
-            Time::new(1, TimeUnit::Millisecond)
-                .to_timezone_aware_string(Some(Timezone::from_tz_string("Europe/Moscow").unwrap()))
+            Time::new(1, TimeUnit::Millisecond).to_timezone_aware_string(Some(
+                &Timezone::from_tz_string("Europe/Moscow").unwrap()
+            ))
         );
     }
 }
diff --git a/src/common/time/src/timestamp.rs b/src/common/time/src/timestamp.rs
index d09890544b..2f13b09a15 100644
--- a/src/common/time/src/timestamp.rs
+++ b/src/common/time/src/timestamp.rs
@@ -14,7 +14,7 @@
 
 use core::default::Default;
 use std::cmp::Ordering;
-use std::fmt::{Display, Formatter};
+use std::fmt::{Display, Formatter, Write};
 use std::hash::{Hash, Hasher};
 use std::str::FromStr;
 use std::time::Duration;
@@ -26,7 +26,9 @@ use chrono::{
 use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt};
 
-use crate::error::{ArithmeticOverflowSnafu, Error, ParseTimestampSnafu, TimestampOverflowSnafu};
+use crate::error::{
+    ArithmeticOverflowSnafu, Error, ParseTimestampSnafu, Result, TimestampOverflowSnafu,
+};
 use crate::timezone::{get_timezone, Timezone};
 use crate::util::div_ceil;
 use crate::{error, Interval};
@@ -290,32 +292,50 @@ impl Timestamp {
     /// Format timestamp to ISO8601 string. If the timestamp exceeds what chrono timestamp can
     /// represent, this function simply print the timestamp unit and value in plain string.
     pub fn to_iso8601_string(&self) -> String {
+        // Safety: the format is valid
         self.as_formatted_string("%Y-%m-%d %H:%M:%S%.f%z", None)
+            .unwrap()
     }
 
     /// Format timestamp use **system timezone**.
     pub fn to_local_string(&self) -> String {
+        // Safety: the format is valid
         self.as_formatted_string("%Y-%m-%d %H:%M:%S%.f", None)
+            .unwrap()
     }
 
     /// Format timestamp for given timezone.
     /// If `tz==None`, the server default timezone will used.
-    pub fn to_timezone_aware_string(&self, tz: Option<Timezone>) -> String {
+    pub fn to_timezone_aware_string(&self, tz: Option<&Timezone>) -> String {
+        // Safety: the format is valid
         self.as_formatted_string("%Y-%m-%d %H:%M:%S%.f", tz)
+            .unwrap()
     }
 
-    fn as_formatted_string(self, pattern: &str, timezone: Option<Timezone>) -> String {
+    /// Format timestamp for given format and timezone.
+    /// If `tz==None`, the server default timezone will used.
+    pub fn as_formatted_string(self, pattern: &str, timezone: Option<&Timezone>) -> Result<String> {
         if let Some(v) = self.to_chrono_datetime() {
+            let mut formatted = String::new();
+
             match get_timezone(timezone) {
                 Timezone::Offset(offset) => {
-                    format!("{}", offset.from_utc_datetime(&v).format(pattern))
+                    write!(
+                        formatted,
+                        "{}",
+                        offset.from_utc_datetime(&v).format(pattern)
+                    )
+                    .context(crate::error::FormatSnafu { pattern })?;
                 }
                 Timezone::Named(tz) => {
-                    format!("{}", tz.from_utc_datetime(&v).format(pattern))
+                    write!(formatted, "{}", tz.from_utc_datetime(&v).format(pattern))
+                        .context(crate::error::FormatSnafu { pattern })?;
                 }
             }
+
+            Ok(formatted)
         } else {
-            format!("[Timestamp{}: {}]", self.unit, self.value)
+            Ok(format!("[Timestamp{}: {}]", self.unit, self.value))
         }
     }
 
@@ -324,7 +344,7 @@ impl Timestamp {
         NaiveDateTime::from_timestamp_opt(sec, nsec)
     }
 
-    pub fn to_chrono_datetime_with_timezone(&self, tz: Option<Timezone>) -> Option<NaiveDateTime> {
+    pub fn to_chrono_datetime_with_timezone(&self, tz: Option<&Timezone>) -> Option<NaiveDateTime> {
         let datetime = self.to_chrono_datetime();
         datetime.map(|v| match tz {
             Some(Timezone::Offset(offset)) => offset.from_utc_datetime(&v).naive_local(),
@@ -369,7 +389,7 @@ impl FromStr for Timestamp {
     /// - `2022-09-20 14:16:43` (Zulu timezone, without T)
     /// - `2022-09-20 14:16:43.012345` (Zulu timezone, without T)
     #[allow(deprecated)]
-    fn from_str(s: &str) -> Result<Self, Self::Err> {
+    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
         // RFC3339 timestamp (with a T)
         let s = s.trim();
         if let Ok(ts) = DateTime::parse_from_rfc3339(s) {
@@ -1113,47 +1133,77 @@ mod tests {
         assert_eq!(
             "1970-01-01 08:00:00.001",
             Timestamp::new(1, TimeUnit::Millisecond)
-                .to_timezone_aware_string(Some(Timezone::from_tz_string("SYSTEM").unwrap()))
+                .to_timezone_aware_string(Some(&Timezone::from_tz_string("SYSTEM").unwrap()))
         );
         assert_eq!(
             "1970-01-01 08:00:00.001",
             Timestamp::new(1, TimeUnit::Millisecond)
-                .to_timezone_aware_string(Some(Timezone::from_tz_string("SYSTEM").unwrap()))
+                .to_timezone_aware_string(Some(&Timezone::from_tz_string("SYSTEM").unwrap()))
         );
         assert_eq!(
             "1970-01-01 08:00:00.001",
             Timestamp::new(1, TimeUnit::Millisecond)
-                .to_timezone_aware_string(Some(Timezone::from_tz_string("+08:00").unwrap()))
+                .to_timezone_aware_string(Some(&Timezone::from_tz_string("+08:00").unwrap()))
         );
         assert_eq!(
             "1970-01-01 07:00:00.001",
             Timestamp::new(1, TimeUnit::Millisecond)
-                .to_timezone_aware_string(Some(Timezone::from_tz_string("+07:00").unwrap()))
+                .to_timezone_aware_string(Some(&Timezone::from_tz_string("+07:00").unwrap()))
         );
         assert_eq!(
             "1969-12-31 23:00:00.001",
             Timestamp::new(1, TimeUnit::Millisecond)
-                .to_timezone_aware_string(Some(Timezone::from_tz_string("-01:00").unwrap()))
+                .to_timezone_aware_string(Some(&Timezone::from_tz_string("-01:00").unwrap()))
         );
         assert_eq!(
             "1970-01-01 08:00:00.001",
-            Timestamp::new(1, TimeUnit::Millisecond)
-                .to_timezone_aware_string(Some(Timezone::from_tz_string("Asia/Shanghai").unwrap()))
+            Timestamp::new(1, TimeUnit::Millisecond).to_timezone_aware_string(Some(
+                &Timezone::from_tz_string("Asia/Shanghai").unwrap()
+            ))
         );
         assert_eq!(
             "1970-01-01 00:00:00.001",
             Timestamp::new(1, TimeUnit::Millisecond)
-                .to_timezone_aware_string(Some(Timezone::from_tz_string("UTC").unwrap()))
+                .to_timezone_aware_string(Some(&Timezone::from_tz_string("UTC").unwrap()))
         );
         assert_eq!(
             "1970-01-01 01:00:00.001",
-            Timestamp::new(1, TimeUnit::Millisecond)
-                .to_timezone_aware_string(Some(Timezone::from_tz_string("Europe/Berlin").unwrap()))
+            Timestamp::new(1, TimeUnit::Millisecond).to_timezone_aware_string(Some(
+                &Timezone::from_tz_string("Europe/Berlin").unwrap()
+            ))
         );
         assert_eq!(
             "1970-01-01 03:00:00.001",
-            Timestamp::new(1, TimeUnit::Millisecond)
-                .to_timezone_aware_string(Some(Timezone::from_tz_string("Europe/Moscow").unwrap()))
+            Timestamp::new(1, TimeUnit::Millisecond).to_timezone_aware_string(Some(
+                &Timezone::from_tz_string("Europe/Moscow").unwrap()
+            ))
+        );
+    }
+
+    #[test]
+    fn test_as_formatted_string() {
+        let ts = Timestamp::new(1, TimeUnit::Millisecond);
+
+        assert_eq!(
+            "1970-01-01",
+            ts.as_formatted_string("%Y-%m-%d", None).unwrap()
+        );
+        assert_eq!(
+            "1970-01-01 00:00:00",
+            ts.as_formatted_string("%Y-%m-%d %H:%M:%S", None).unwrap()
+        );
+        assert_eq!(
+            "1970-01-01T00:00:00:001",
+            ts.as_formatted_string("%Y-%m-%dT%H:%M:%S:%3f", None)
+                .unwrap()
+        );
+        assert_eq!(
+            "1970-01-01T08:00:00:001",
+            ts.as_formatted_string(
+                "%Y-%m-%dT%H:%M:%S:%3f",
+                Some(&Timezone::from_tz_string("Asia/Shanghai").unwrap())
+            )
+            .unwrap()
         );
     }
 
diff --git a/src/common/time/src/timezone.rs b/src/common/time/src/timezone.rs
index 0e59708494..dda94eb684 100644
--- a/src/common/time/src/timezone.rs
+++ b/src/common/time/src/timezone.rs
@@ -43,13 +43,8 @@ pub fn set_default_timezone(tz_str: Option<&str>) -> Result<()> {
 #[inline(always)]
 /// If the `tz=Some(timezone)`, return `timezone` directly,
 /// or return current system timezone.
-pub fn get_timezone(tz: Option<Timezone>) -> Timezone {
-    tz.unwrap_or_else(|| {
-        DEFAULT_TIMEZONE
-            .get()
-            .cloned()
-            .unwrap_or(Timezone::Named(Tz::UTC))
-    })
+pub fn get_timezone(tz: Option<&Timezone>) -> &Timezone {
+    tz.unwrap_or_else(|| DEFAULT_TIMEZONE.get().unwrap_or(&Timezone::Named(Tz::UTC)))
 }
 
 #[inline(always)]
diff --git a/src/datatypes/src/value.rs b/src/datatypes/src/value.rs
index 943f563a0a..fede1420bb 100644
--- a/src/datatypes/src/value.rs
+++ b/src/datatypes/src/value.rs
@@ -218,6 +218,14 @@ impl Value {
         }
     }
 
+    /// Cast Value to utf8 String. Return None if value is not a valid string data type.
+    pub fn as_string(&self) -> Option<String> {
+        match self {
+            Value::String(bytes) => Some(bytes.as_utf8().to_string()),
+            _ => None,
+        }
+    }
+
     /// Cast Value to Date. Return None if value is not a valid date data type.
     pub fn as_date(&self) -> Option<Date> {
         match self {
diff --git a/src/servers/src/mysql/writer.rs b/src/servers/src/mysql/writer.rs
index 726daf061f..fd5304ec3e 100644
--- a/src/servers/src/mysql/writer.rs
+++ b/src/servers/src/mysql/writer.rs
@@ -194,10 +194,10 @@ impl<'a, W: AsyncWrite + Unpin> MysqlResultWriter<'a, W> {
                     Value::Date(v) => row_writer.write_col(v.to_chrono_date())?,
                     // convert datetime and timestamp to timezone of current connection
                     Value::DateTime(v) => row_writer.write_col(
-                        v.to_chrono_datetime_with_timezone(Some(query_context.timezone())),
+                        v.to_chrono_datetime_with_timezone(Some(&query_context.timezone())),
                     )?,
                     Value::Timestamp(v) => row_writer.write_col(
-                        v.to_chrono_datetime_with_timezone(Some(query_context.timezone())),
+                        v.to_chrono_datetime_with_timezone(Some(&query_context.timezone())),
                     )?,
                     Value::Interval(v) => row_writer.write_col(v.to_iso8601_string())?,
                     Value::Duration(v) => row_writer.write_col(v.to_std_duration())?,
@@ -210,7 +210,7 @@ impl<'a, W: AsyncWrite + Unpin> MysqlResultWriter<'a, W> {
                         })
                     }
                     Value::Time(v) => row_writer
-                        .write_col(v.to_timezone_aware_string(Some(query_context.timezone())))?,
+                        .write_col(v.to_timezone_aware_string(Some(&query_context.timezone())))?,
                     Value::Decimal128(v) => row_writer.write_col(v.to_string())?,
                 }
             }
diff --git a/src/session/src/context.rs b/src/session/src/context.rs
index 7446624ae5..4e3314215c 100644
--- a/src/session/src/context.rs
+++ b/src/session/src/context.rs
@@ -69,7 +69,7 @@ impl From<&RegionRequestHeader> for QueryContext {
             current_schema: schema.to_string(),
             current_user: Default::default(),
             // for request send to datanode, all timestamp have converted to UTC, so timezone is not important
-            timezone: ArcSwap::new(Arc::new(get_timezone(None))),
+            timezone: ArcSwap::new(Arc::new(get_timezone(None).clone())),
             sql_dialect: Box::new(GreptimeDbDialect {}),
         }
     }
@@ -123,8 +123,8 @@ impl QueryContext {
         build_db_string(catalog, schema)
     }
 
-    pub fn timezone(&self) -> Timezone {
-        self.timezone.load().as_ref().clone()
+    pub fn timezone(&self) -> Arc<Timezone> {
+        self.timezone.load().clone()
     }
 
     pub fn current_user(&self) -> Option<UserInfoRef> {
@@ -143,8 +143,8 @@ impl QueryContext {
     /// We need persist these change in `Session`.
     pub fn update_session(&self, session: &SessionRef) {
         let tz = self.timezone();
-        if session.timezone() != tz {
-            session.set_timezone(tz)
+        if session.timezone() != *tz {
+            session.set_timezone(tz.as_ref().clone())
         }
     }
 }
@@ -163,7 +163,7 @@ impl QueryContextBuilder {
                 .unwrap_or_else(|| ArcSwap::new(Arc::new(None))),
             timezone: self
                 .timezone
-                .unwrap_or(ArcSwap::new(Arc::new(get_timezone(None)))),
+                .unwrap_or(ArcSwap::new(Arc::new(get_timezone(None).clone()))),
             sql_dialect: self
                 .sql_dialect
                 .unwrap_or_else(|| Box::new(GreptimeDbDialect {})),
diff --git a/src/session/src/lib.rs b/src/session/src/lib.rs
index 35035cda27..f64ac81885 100644
--- a/src/session/src/lib.rs
+++ b/src/session/src/lib.rs
@@ -46,7 +46,7 @@ impl Session {
             schema: ArcSwap::new(Arc::new(DEFAULT_SCHEMA_NAME.into())),
             user_info: ArcSwap::new(Arc::new(auth::userinfo_by_name(None))),
             conn_info: ConnInfo::new(addr, channel),
-            timezone: ArcSwap::new(Arc::new(get_timezone(None))),
+            timezone: ArcSwap::new(Arc::new(get_timezone(None).clone())),
         }
     }
 
diff --git a/tests/cases/standalone/common/function/date.result b/tests/cases/standalone/common/function/date.result
index 122866efac..a94cff3cd3 100644
--- a/tests/cases/standalone/common/function/date.result
+++ b/tests/cases/standalone/common/function/date.result
@@ -1,3 +1,4 @@
+--- date_add ---
 SELECT date_add('2023-12-06 07:39:46.222'::TIMESTAMP_MS, INTERVAL '5 day');
 
 +----------------------------------------------------------------------------------------+
@@ -30,6 +31,7 @@ SELECT date_add('2023-12-06'::DATE, '3 month 5 day');
 | 2024-03-11                                         |
 +----------------------------------------------------+
 
+--- date_sub ---
 SELECT date_sub('2023-12-06 07:39:46.222'::TIMESTAMP_MS, INTERVAL '5 day');
 
 +----------------------------------------------------------------------------------------+
@@ -62,6 +64,37 @@ SELECT date_sub('2023-12-06'::DATE, '3 month 5 day');
 | 2023-09-01                                         |
 +----------------------------------------------------+
 
+--- date_format ---
+SELECT date_format('2023-12-06 07:39:46.222'::TIMESTAMP_MS, '%Y-%m-%d %H:%M:%S:%3f');
+
++----------------------------------------------------------------------------+
+| date_format(Utf8("2023-12-06 07:39:46.222"),Utf8("%Y-%m-%d %H:%M:%S:%3f")) |
++----------------------------------------------------------------------------+
+| 2023-12-06 07:39:46:222                                                    |
++----------------------------------------------------------------------------+
+
+SELECT date_format('2023-12-06 07:39:46.222'::TIMESTAMP_S, '%Y-%m-%d %H:%M:%S:%3f');
+
++----------------------------------------------------------------------------+
+| date_format(Utf8("2023-12-06 07:39:46.222"),Utf8("%Y-%m-%d %H:%M:%S:%3f")) |
++----------------------------------------------------------------------------+
+| 2023-12-06 07:39:46:000                                                    |
++----------------------------------------------------------------------------+
+
+--- datetime not supported yet ---
+SELECT date_format('2023-12-06 07:39:46.222'::DATETIME, '%Y-%m-%d %H:%M:%S:%3f');
+
+Error: 3000(PlanQuery), Failed to plan SQL: This feature is not implemented: Unsupported SQL type Datetime(None)
+
+SELECT date_format('2023-12-06'::DATE, '%m-%d');
+
++-----------------------------------------------+
+| date_format(Utf8("2023-12-06"),Utf8("%m-%d")) |
++-----------------------------------------------+
+| 12-06                                         |
++-----------------------------------------------+
+
+--- test date functions with table rows ---
 CREATE TABLE dates(d DATE, ts timestamp time index);
 
 Affected Rows: 0
@@ -78,6 +111,7 @@ INSERT INTO dates VALUES ('2023-12-06'::DATE, 3);
 
 Affected Rows: 1
 
+--- date_add ---
 SELECT date_add(d, INTERVAL '1 year 2 month 3 day') from dates;
 
 +---------------------------------------------------------------------------+
@@ -118,6 +152,7 @@ SELECT date_add(ts, '1 year 2 month 3 day') from dates;
 | 1971-03-04T00:00:00.003                         |
 +-------------------------------------------------+
 
+--- date_sub ---
 SELECT date_sub(d, INTERVAL '1 year 2 month 3 day') from dates;
 
 +---------------------------------------------------------------------------+
@@ -158,6 +193,27 @@ SELECT date_sub(ts, '1 year 2 month 3 day') from dates;
 | 1968-10-29T00:00:00.003                         |
 +-------------------------------------------------+
 
+--- date_format ---
+SELECT date_format(d, '%Y-%m-%d %H:%M:%S:%3f') from dates;
+
++----------------------------------------------------+
+| date_format(dates.d,Utf8("%Y-%m-%d %H:%M:%S:%3f")) |
++----------------------------------------------------+
+| 1992-01-01 00:00:00:000                            |
+| 1993-12-30 00:00:00:000                            |
+| 2023-12-06 00:00:00:000                            |
++----------------------------------------------------+
+
+SELECT date_format(ts, '%Y-%m-%d %H:%M:%S:%3f') from dates;
+
++-----------------------------------------------------+
+| date_format(dates.ts,Utf8("%Y-%m-%d %H:%M:%S:%3f")) |
++-----------------------------------------------------+
+| 1970-01-01 00:00:00:001                             |
+| 1970-01-01 00:00:00:002                             |
+| 1970-01-01 00:00:00:003                             |
++-----------------------------------------------------+
+
 DROP TABLE dates;
 
 Affected Rows: 0
diff --git a/tests/cases/standalone/common/function/date.sql b/tests/cases/standalone/common/function/date.sql
index a5f9627499..db661ca4ed 100644
--- a/tests/cases/standalone/common/function/date.sql
+++ b/tests/cases/standalone/common/function/date.sql
@@ -1,3 +1,4 @@
+--- date_add ---
 SELECT date_add('2023-12-06 07:39:46.222'::TIMESTAMP_MS, INTERVAL '5 day');
 
 SELECT date_add('2023-12-06 07:39:46.222'::TIMESTAMP_MS, '5 day');
@@ -6,6 +7,7 @@ SELECT date_add('2023-12-06'::DATE, INTERVAL '3 month 5 day');
 
 SELECT date_add('2023-12-06'::DATE, '3 month 5 day');
 
+--- date_sub ---
 SELECT date_sub('2023-12-06 07:39:46.222'::TIMESTAMP_MS, INTERVAL '5 day');
 
 SELECT date_sub('2023-12-06 07:39:46.222'::TIMESTAMP_MS, '5 day');
@@ -14,7 +16,17 @@ SELECT date_sub('2023-12-06'::DATE, INTERVAL '3 month 5 day');
 
 SELECT date_sub('2023-12-06'::DATE, '3 month 5 day');
 
+--- date_format ---
+SELECT date_format('2023-12-06 07:39:46.222'::TIMESTAMP_MS, '%Y-%m-%d %H:%M:%S:%3f');
 
+SELECT date_format('2023-12-06 07:39:46.222'::TIMESTAMP_S, '%Y-%m-%d %H:%M:%S:%3f');
+
+--- datetime not supported yet ---
+SELECT date_format('2023-12-06 07:39:46.222'::DATETIME, '%Y-%m-%d %H:%M:%S:%3f');
+
+SELECT date_format('2023-12-06'::DATE, '%m-%d');
+
+--- test date functions with table rows ---
 CREATE TABLE dates(d DATE, ts timestamp time index);
 
 INSERT INTO dates VALUES ('1992-01-01'::DATE, 1);
@@ -23,6 +35,7 @@ INSERT INTO dates VALUES ('1993-12-30'::DATE, 2);
 
 INSERT INTO dates VALUES ('2023-12-06'::DATE, 3);
 
+--- date_add ---
 SELECT date_add(d, INTERVAL '1 year 2 month 3 day') from dates;
 
 SELECT date_add(d, '1 year 2 month 3 day') from dates;
@@ -31,6 +44,7 @@ SELECT date_add(ts, INTERVAL '1 year 2 month 3 day') from dates;
 
 SELECT date_add(ts, '1 year 2 month 3 day') from dates;
 
+--- date_sub ---
 SELECT date_sub(d, INTERVAL '1 year 2 month 3 day') from dates;
 
 SELECT date_sub(d, '1 year 2 month 3 day') from dates;
@@ -39,4 +53,9 @@ SELECT date_sub(ts, INTERVAL '1 year 2 month 3 day') from dates;
 
 SELECT date_sub(ts, '1 year 2 month 3 day') from dates;
 
+--- date_format ---
+SELECT date_format(d, '%Y-%m-%d %H:%M:%S:%3f') from dates;
+
+SELECT date_format(ts, '%Y-%m-%d %H:%M:%S:%3f') from dates;
+
 DROP TABLE dates;