From eed97cfaa0bd01a6c6da4b3e4baee89a6680adfd Mon Sep 17 00:00:00 2001
From: Siew Kam Onn <kosiew@gmail.com>
Date: Sun, 12 Oct 2025 18:29:43 +0800
Subject: [PATCH 01/11] =?UTF-8?q?Implement=20timezone-aware=20handling=20f?=
 =?UTF-8?q?or=20to=5Ftimestamp=20functions.=20Refactor=20shared=20helpers?=
 =?UTF-8?q?=20to=20ensure=20nai=CC=88ve=20strings=20are=20interpreted=20us?=
 =?UTF-8?q?ing=20the=20configured=20execution=20zone.=20Extend=20unit=20te?=
 =?UTF-8?q?sts=20to=20cover=20nai=CC=88ve=20and=20formatted=20inputs=20res?=
 =?UTF-8?q?pecting=20non-UTC=20execution=20timezones.?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 Cargo.lock                                    |   1 +
 datafusion/functions/Cargo.toml               |   1 +
 datafusion/functions/benches/to_timestamp.rs  |  19 +-
 datafusion/functions/src/datetime/common.rs   | 158 +++-
 datafusion/functions/src/datetime/mod.rs      |  46 +-
 datafusion/functions/src/datetime/to_date.rs  |   6 +-
 .../functions/src/datetime/to_timestamp.rs    | 829 +++++++++++++++---
 .../functions/src/datetime/to_unixtime.rs     |   8 +-
 datafusion/functions/src/macros.rs            |  29 +
 .../sqllogictest/test_files/timestamps.slt    |  12 +-
 .../test_files/to_timestamp_timezone.slt      | 204 +++++
 .../source/user-guide/sql/scalar_functions.md |  79 +-
 12 files changed, 1221 insertions(+), 171 deletions(-)
 create mode 100644 datafusion/sqllogictest/test_files/to_timestamp_timezone.slt

diff --git a/Cargo.lock b/Cargo.lock
index 6415fa65c09e4..aca33b8bbf68f 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2309,6 +2309,7 @@ dependencies = [
  "blake2",
  "blake3",
  "chrono",
+ "chrono-tz",
  "criterion",
  "ctor",
  "datafusion-common",
diff --git a/datafusion/functions/Cargo.toml b/datafusion/functions/Cargo.toml
index a384e5635d8c3..01c78ebc5f8bd 100644
--- a/datafusion/functions/Cargo.toml
+++ b/datafusion/functions/Cargo.toml
@@ -71,6 +71,7 @@ base64 = { version = "0.22", optional = true }
 blake2 = { version = "^0.10.2", optional = true }
 blake3 = { version = "1.8", optional = true }
 chrono = { workspace = true }
+chrono-tz = "0.10.4"
 datafusion-common = { workspace = true }
 datafusion-doc = { workspace = true }
 datafusion-execution = { workspace = true }
diff --git a/datafusion/functions/benches/to_timestamp.rs b/datafusion/functions/benches/to_timestamp.rs
index a8f5c5816d4da..c0803641a6eaf 100644
--- a/datafusion/functions/benches/to_timestamp.rs
+++ b/datafusion/functions/benches/to_timestamp.rs
@@ -115,15 +115,17 @@ fn criterion_benchmark(c: &mut Criterion) {
     let arg_field = Field::new("a", DataType::Utf8, false).into();
     let arg_fields = vec![arg_field];
     let config_options = Arc::new(ConfigOptions::default());
+    let to_timestamp_udf = to_timestamp(config_options.as_ref());
 
     c.bench_function("to_timestamp_no_formats_utf8", |b| {
+        let to_timestamp_udf = Arc::clone(&to_timestamp_udf);
         let arr_data = data();
         let batch_len = arr_data.len();
         let string_array = ColumnarValue::Array(Arc::new(arr_data) as ArrayRef);
 
         b.iter(|| {
             black_box(
-                to_timestamp()
+                to_timestamp_udf
                     .invoke_with_args(ScalarFunctionArgs {
                         args: vec![string_array.clone()],
                         arg_fields: arg_fields.clone(),
@@ -137,13 +139,14 @@ fn criterion_benchmark(c: &mut Criterion) {
     });
 
     c.bench_function("to_timestamp_no_formats_largeutf8", |b| {
+        let to_timestamp_udf = Arc::clone(&to_timestamp_udf);
         let data = cast(&data(), &DataType::LargeUtf8).unwrap();
         let batch_len = data.len();
         let string_array = ColumnarValue::Array(Arc::new(data) as ArrayRef);
 
         b.iter(|| {
             black_box(
-                to_timestamp()
+                to_timestamp_udf
                     .invoke_with_args(ScalarFunctionArgs {
                         args: vec![string_array.clone()],
                         arg_fields: arg_fields.clone(),
@@ -157,13 +160,14 @@ fn criterion_benchmark(c: &mut Criterion) {
     });
 
     c.bench_function("to_timestamp_no_formats_utf8view", |b| {
+        let to_timestamp_udf = Arc::clone(&to_timestamp_udf);
         let data = cast(&data(), &DataType::Utf8View).unwrap();
         let batch_len = data.len();
         let string_array = ColumnarValue::Array(Arc::new(data) as ArrayRef);
 
         b.iter(|| {
             black_box(
-                to_timestamp()
+                to_timestamp_udf
                     .invoke_with_args(ScalarFunctionArgs {
                         args: vec![string_array.clone()],
                         arg_fields: arg_fields.clone(),
@@ -177,6 +181,7 @@ fn criterion_benchmark(c: &mut Criterion) {
     });
 
     c.bench_function("to_timestamp_with_formats_utf8", |b| {
+        let to_timestamp_udf = Arc::clone(&to_timestamp_udf);
         let (inputs, format1, format2, format3) = data_with_formats();
         let batch_len = inputs.len();
 
@@ -196,7 +201,7 @@ fn criterion_benchmark(c: &mut Criterion) {
 
         b.iter(|| {
             black_box(
-                to_timestamp()
+                to_timestamp_udf
                     .invoke_with_args(ScalarFunctionArgs {
                         args: args.clone(),
                         arg_fields: arg_fields.clone(),
@@ -210,6 +215,7 @@ fn criterion_benchmark(c: &mut Criterion) {
     });
 
     c.bench_function("to_timestamp_with_formats_largeutf8", |b| {
+        let to_timestamp_udf = Arc::clone(&to_timestamp_udf);
         let (inputs, format1, format2, format3) = data_with_formats();
         let batch_len = inputs.len();
 
@@ -237,7 +243,7 @@ fn criterion_benchmark(c: &mut Criterion) {
 
         b.iter(|| {
             black_box(
-                to_timestamp()
+                to_timestamp_udf
                     .invoke_with_args(ScalarFunctionArgs {
                         args: args.clone(),
                         arg_fields: arg_fields.clone(),
@@ -251,6 +257,7 @@ fn criterion_benchmark(c: &mut Criterion) {
     });
 
     c.bench_function("to_timestamp_with_formats_utf8view", |b| {
+        let to_timestamp_udf = Arc::clone(&to_timestamp_udf);
         let (inputs, format1, format2, format3) = data_with_formats();
 
         let batch_len = inputs.len();
@@ -279,7 +286,7 @@ fn criterion_benchmark(c: &mut Criterion) {
 
         b.iter(|| {
             black_box(
-                to_timestamp()
+                to_timestamp_udf
                     .invoke_with_args(ScalarFunctionArgs {
                         args: args.clone(),
                         arg_fields: arg_fields.clone(),
diff --git a/datafusion/functions/src/datetime/common.rs b/datafusion/functions/src/datetime/common.rs
index a0daab66c7af3..95e08717d1abe 100644
--- a/datafusion/functions/src/datetime/common.rs
+++ b/datafusion/functions/src/datetime/common.rs
@@ -15,33 +15,52 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::marker::PhantomData;
 use std::sync::Arc;
 
+use arrow::array::timezone::Tz;
 use arrow::array::{
     Array, ArrowPrimitiveType, AsArray, GenericStringArray, PrimitiveArray,
     StringArrayType, StringViewArray,
 };
-use arrow::compute::kernels::cast_utils::string_to_timestamp_nanos;
-use arrow::datatypes::DataType;
+use arrow::compute::kernels::cast_utils::{
+    string_to_datetime, string_to_timestamp_nanos,
+};
+use arrow::datatypes::{DataType, TimeUnit};
+use arrow_buffer::ArrowNativeType;
 use chrono::format::{parse, Parsed, StrftimeItems};
 use chrono::LocalResult::Single;
 use chrono::{DateTime, TimeZone, Utc};
-
 use datafusion_common::cast::as_generic_string_array;
 use datafusion_common::{
-    exec_datafusion_err, exec_err, unwrap_or_internal_err, DataFusionError, Result,
-    ScalarType, ScalarValue,
+    exec_datafusion_err, exec_err, internal_datafusion_err, unwrap_or_internal_err,
+    DataFusionError, Result, ScalarValue,
 };
 use datafusion_expr::ColumnarValue;
+use num_traits::{PrimInt, ToPrimitive};
 
 /// Error message if nanosecond conversion request beyond supported interval
 const ERR_NANOSECONDS_NOT_SUPPORTED: &str = "The dates that can be represented as nanoseconds have to be between 1677-09-21T00:12:44.0 and 2262-04-11T23:47:16.854775804";
 
+#[expect(unused)]
 /// Calls string_to_timestamp_nanos and converts the error type
 pub(crate) fn string_to_timestamp_nanos_shim(s: &str) -> Result<i64> {
     string_to_timestamp_nanos(s).map_err(|e| e.into())
 }
 
+pub(crate) fn string_to_timestamp_nanos_with_timezone(
+    timezone: &Option<Tz>,
+    s: &str,
+) -> Result<i64> {
+    let tz = timezone.unwrap_or("UTC".parse()?);
+    let dt = string_to_datetime(&tz, s)?;
+    let parsed = dt
+        .timestamp_nanos_opt()
+        .ok_or_else(|| exec_datafusion_err!("{ERR_NANOSECONDS_NOT_SUPPORTED}"))?;
+
+    Ok(parsed)
+}
+
 /// Checks that all the arguments from the second are of type [Utf8], [LargeUtf8] or [Utf8View]
 ///
 /// [Utf8]: DataType::Utf8
@@ -69,13 +88,12 @@ pub(crate) fn validate_data_types(args: &[ColumnarValue], name: &str) -> Result<
 /// Accepts a string and parses it using the [`chrono::format::strftime`] specifiers
 /// relative to the provided `timezone`
 ///
-/// [IANA timezones] are only supported if the `arrow-array/chrono-tz` feature is enabled
-///
-/// * `2023-01-01 040506 America/Los_Angeles`
-///
 /// If a timestamp is ambiguous, for example as a result of daylight-savings time, an error
 /// will be returned
 ///
+/// Note that parsing [IANA timezones] is not supported yet in chrono - <https://github.com/chronotope/chrono/issues/38>
+/// and this implementation only supports named timezones at the end of the string preceded by a space.
+///
 /// [`chrono::format::strftime`]: https://docs.rs/chrono/latest/chrono/format/strftime/index.html
 /// [IANA timezones]: https://www.iana.org/time-zones
 pub(crate) fn string_to_datetime_formatted<T: TimeZone>(
@@ -89,11 +107,52 @@ pub(crate) fn string_to_datetime_formatted<T: TimeZone>(
         )
     };
 
+    let mut datetime_str = s;
+    let mut format = format;
+
+    // we manually handle the most common case of a named timezone at the end of the timestamp
+    // not that %+ handles 'Z' at the end of the string without a space. This code doesn't
+    // handle named timezones with no preceding space since that would require writing a
+    // custom parser (or switching to Jiff)
+    let tz: Option<chrono_tz::Tz> = if format.ends_with(" %Z") {
+        // grab the string after the last space as the named timezone
+        let parts: Vec<&str> = datetime_str.rsplitn(2, ' ').collect();
+        let timezone_name = parts[0];
+        datetime_str = parts[1];
+
+        // attempt to parse the timezone name
+        let result: Result<chrono_tz::Tz, chrono_tz::ParseError> = timezone_name.parse();
+        let Ok(tz) = result else {
+            return Err(err(&result.unwrap_err().to_string()));
+        };
+
+        // successfully parsed the timezone name, remove the ' %Z' from the format
+        format = format.trim_end_matches(" %Z");
+
+        Some(tz)
+    } else if format.contains("%Z") {
+        return Err(err(
+            "'%Z' is only supported at the end of the format string preceded by a space",
+        ));
+    } else {
+        None
+    };
+
     let mut parsed = Parsed::new();
-    parse(&mut parsed, s, StrftimeItems::new(format)).map_err(|e| err(&e.to_string()))?;
+    parse(&mut parsed, datetime_str, StrftimeItems::new(format))
+        .map_err(|e| err(&e.to_string()))?;
 
-    // attempt to parse the string assuming it has a timezone
-    let dt = parsed.to_datetime();
+    let dt = match tz {
+        Some(tz) => {
+            // A timezone was manually parsed out, convert it to a fixed offset
+            match parsed.to_datetime_with_timezone(&tz) {
+                Ok(dt) => Ok(dt.fixed_offset()),
+                Err(e) => Err(e),
+            }
+        }
+        // default to parse the string assuming it has a timezone
+        None => parsed.to_datetime(),
+    };
 
     if let Err(e) = &dt {
         // no timezone or other failure, try without a timezone
@@ -141,6 +200,7 @@ pub(crate) fn string_to_datetime_formatted<T: TimeZone>(
 ///
 /// [`chrono::format::strftime`]: https://docs.rs/chrono/latest/chrono/format/strftime/index.html
 #[inline]
+#[expect(unused)]
 pub(crate) fn string_to_timestamp_nanos_formatted(
     s: &str,
     format: &str,
@@ -152,6 +212,20 @@ pub(crate) fn string_to_timestamp_nanos_formatted(
         .ok_or_else(|| exec_datafusion_err!("{ERR_NANOSECONDS_NOT_SUPPORTED}"))
 }
 
+pub(crate) fn string_to_timestamp_nanos_formatted_with_timezone(
+    timezone: &Option<Tz>,
+    s: &str,
+    format: &str,
+) -> Result<i64, DataFusionError> {
+    let dt =
+        string_to_datetime_formatted(&timezone.unwrap_or("UTC".parse()?), s, format)?;
+    let parsed = dt
+        .timestamp_nanos_opt()
+        .ok_or_else(|| exec_datafusion_err!("{ERR_NANOSECONDS_NOT_SUPPORTED}"))?;
+
+    Ok(parsed)
+}
+
 /// Accepts a string with a `chrono` format and converts it to a
 /// millisecond precision timestamp.
 ///
@@ -176,14 +250,50 @@ pub(crate) fn string_to_timestamp_millis_formatted(s: &str, format: &str) -> Res
         .timestamp_millis())
 }
 
-pub(crate) fn handle<O, F, S>(
+pub(crate) struct ScalarDataType<T: PrimInt> {
+    data_type: DataType,
+    _marker: PhantomData<T>,
+}
+
+impl<T: PrimInt> ScalarDataType<T> {
+    pub(crate) fn new(dt: DataType) -> Self {
+        Self {
+            data_type: dt,
+            _marker: PhantomData,
+        }
+    }
+
+    fn scalar(&self, r: Option<i64>) -> Result<ScalarValue> {
+        match &self.data_type {
+            DataType::Date32 => Ok(ScalarValue::Date32(r.and_then(|v| v.to_i32()))),
+            DataType::Timestamp(u, tz) => match u {
+                TimeUnit::Second => Ok(ScalarValue::TimestampSecond(r, tz.clone())),
+                TimeUnit::Millisecond => {
+                    Ok(ScalarValue::TimestampMillisecond(r, tz.clone()))
+                }
+                TimeUnit::Microsecond => {
+                    Ok(ScalarValue::TimestampMicrosecond(r, tz.clone()))
+                }
+                TimeUnit::Nanosecond => {
+                    Ok(ScalarValue::TimestampNanosecond(r, tz.clone()))
+                }
+            },
+            t => Err(internal_datafusion_err!(
+                "Unsupported data type for ScalarDataType<T>: {t:?}"
+            )),
+        }
+    }
+}
+
+pub(crate) fn handle<O, F, T>(
     args: &[ColumnarValue],
     op: F,
     name: &str,
+    sdt: &ScalarDataType<T>,
 ) -> Result<ColumnarValue>
 where
     O: ArrowPrimitiveType,
-    S: ScalarType<O::Native>,
+    T: PrimInt,
     F: Fn(&str) -> Result<O::Native>,
 {
     match &args[0] {
@@ -210,8 +320,13 @@ where
         },
         ColumnarValue::Scalar(scalar) => match scalar.try_as_str() {
             Some(a) => {
-                let result = a.as_ref().map(|x| op(x)).transpose()?;
-                Ok(ColumnarValue::Scalar(S::scalar(result)))
+                let result = a
+                    .as_ref()
+                    .map(|x| op(x))
+                    .transpose()?
+                    .and_then(|v| v.to_i64());
+                let s = sdt.scalar(result)?;
+                Ok(ColumnarValue::Scalar(s))
             }
             _ => exec_err!("Unsupported data type {scalar:?} for function {name}"),
         },
@@ -221,17 +336,18 @@ where
 // Given a function that maps a `&str`, `&str` to an arrow native type,
 // returns a `ColumnarValue` where the function is applied to either a `ArrayRef` or `ScalarValue`
 // depending on the `args`'s variant.
-pub(crate) fn handle_multiple<O, F, S, M>(
+pub(crate) fn handle_multiple<O, F, M, T>(
     args: &[ColumnarValue],
     op: F,
     op2: M,
     name: &str,
+    sdt: &ScalarDataType<T>,
 ) -> Result<ColumnarValue>
 where
     O: ArrowPrimitiveType,
-    S: ScalarType<O::Native>,
     F: Fn(&str, &str) -> Result<O::Native>,
     M: Fn(O::Native) -> O::Native,
+    T: PrimInt,
 {
     match &args[0] {
         ColumnarValue::Array(a) => match a.data_type() {
@@ -286,9 +402,9 @@ where
                     if let Some(s) = x {
                         match op(a, s.as_str()) {
                             Ok(r) => {
-                                ret = Some(Ok(ColumnarValue::Scalar(S::scalar(Some(
-                                    op2(r),
-                                )))));
+                                let result = op2(r).to_i64();
+                                let s = sdt.scalar(result)?;
+                                ret = Some(Ok(ColumnarValue::Scalar(s)));
                                 break;
                             }
                             Err(e) => ret = Some(Err(e)),
diff --git a/datafusion/functions/src/datetime/mod.rs b/datafusion/functions/src/datetime/mod.rs
index d80f14facf822..a6a04cde923bd 100644
--- a/datafusion/functions/src/datetime/mod.rs
+++ b/datafusion/functions/src/datetime/mod.rs
@@ -49,11 +49,14 @@ make_udf_function!(to_char::ToCharFunc, to_char);
 make_udf_function!(to_date::ToDateFunc, to_date);
 make_udf_function!(to_local_time::ToLocalTimeFunc, to_local_time);
 make_udf_function!(to_unixtime::ToUnixtimeFunc, to_unixtime);
-make_udf_function!(to_timestamp::ToTimestampFunc, to_timestamp);
-make_udf_function!(to_timestamp::ToTimestampSecondsFunc, to_timestamp_seconds);
-make_udf_function!(to_timestamp::ToTimestampMillisFunc, to_timestamp_millis);
-make_udf_function!(to_timestamp::ToTimestampMicrosFunc, to_timestamp_micros);
-make_udf_function!(to_timestamp::ToTimestampNanosFunc, to_timestamp_nanos);
+make_udf_function_with_config!(to_timestamp::ToTimestampFunc, to_timestamp);
+make_udf_function_with_config!(
+    to_timestamp::ToTimestampSecondsFunc,
+    to_timestamp_seconds
+);
+make_udf_function_with_config!(to_timestamp::ToTimestampMillisFunc, to_timestamp_millis);
+make_udf_function_with_config!(to_timestamp::ToTimestampMicrosFunc, to_timestamp_micros);
+make_udf_function_with_config!(to_timestamp::ToTimestampNanosFunc, to_timestamp_nanos);
 
 // create UDF with config
 make_udf_function_with_config!(now::NowFunc, now);
@@ -106,24 +109,24 @@ pub mod expr_fn {
         args,
     ),(
         to_timestamp,
-        "converts a string and optional formats to a `Timestamp(Nanoseconds, None)`",
-        args,
+        "converts a string and optional formats to a `Timestamp(Nanoseconds, TimeZone)`",
+        @config args,
     ),(
         to_timestamp_seconds,
-        "converts a string and optional formats to a `Timestamp(Seconds, None)`",
-        args,
+        "converts a string and optional formats to a `Timestamp(Seconds, TimeZone)`",
+        @config args,
     ),(
         to_timestamp_millis,
-        "converts a string and optional formats to a `Timestamp(Milliseconds, None)`",
-        args,
+        "converts a string and optional formats to a `Timestamp(Milliseconds, TimeZone)`",
+        @config args,
     ),(
         to_timestamp_micros,
-        "converts a string and optional formats to a `Timestamp(Microseconds, None)`",
-        args,
+        "converts a string and optional formats to a `Timestamp(Microseconds, TimeZone)`",
+        @config args,
     ),(
         to_timestamp_nanos,
-        "converts a string and optional formats to a `Timestamp(Nanoseconds, None)`",
-        args,
+        "converts a string and optional formats to a `Timestamp(Nanoseconds, TimeZone)`",
+        @config args,
     ));
 
     /// Returns a string representation of a date, time, timestamp or duration based
@@ -259,6 +262,7 @@ pub mod expr_fn {
 /// Returns all DataFusion functions defined in this package
 pub fn functions() -> Vec<Arc<ScalarUDF>> {
     use datafusion_common::config::ConfigOptions;
+    let config = ConfigOptions::default();
     vec![
         current_date(),
         current_time(),
@@ -267,15 +271,15 @@ pub fn functions() -> Vec<Arc<ScalarUDF>> {
         date_trunc(),
         from_unixtime(),
         make_date(),
-        now(&ConfigOptions::default()),
+        now(&config),
         to_char(),
         to_date(),
         to_local_time(),
         to_unixtime(),
-        to_timestamp(),
-        to_timestamp_seconds(),
-        to_timestamp_millis(),
-        to_timestamp_micros(),
-        to_timestamp_nanos(),
+        to_timestamp(&config),
+        to_timestamp_seconds(&config),
+        to_timestamp_millis(&config),
+        to_timestamp_micros(&config),
+        to_timestamp_nanos(&config),
     ]
 }
diff --git a/datafusion/functions/src/datetime/to_date.rs b/datafusion/functions/src/datetime/to_date.rs
index a216e8e5b74ac..746b90e5338a9 100644
--- a/datafusion/functions/src/datetime/to_date.rs
+++ b/datafusion/functions/src/datetime/to_date.rs
@@ -82,7 +82,7 @@ impl ToDateFunc {
 
     fn to_date(&self, args: &[ColumnarValue]) -> Result<ColumnarValue> {
         match args.len() {
-            1 => handle::<Date32Type, _, Date32Type>(
+            1 => handle::<Date32Type, _, i32>(
                 args,
                 |s| match Date32Type::parse(s) {
                     Some(v) => Ok(v),
@@ -92,8 +92,9 @@ impl ToDateFunc {
                     )),
                 },
                 "to_date",
+                &ScalarDataType::new(Date32),
             ),
-            2.. => handle_multiple::<Date32Type, _, Date32Type, _>(
+            2.. => handle_multiple::<Date32Type, _, _, i32>(
                 args,
                 |s, format| {
                     string_to_timestamp_millis_formatted(s, format)
@@ -106,6 +107,7 @@ impl ToDateFunc {
                 },
                 |n| n,
                 "to_date",
+                &ScalarDataType::new(Date32),
             ),
             0 => exec_err!("Unsupported 0 argument count for function to_date"),
         }
diff --git a/datafusion/functions/src/datetime/to_timestamp.rs b/datafusion/functions/src/datetime/to_timestamp.rs
index cf6c3429387c4..ce5ed3a4a92a4 100644
--- a/datafusion/functions/src/datetime/to_timestamp.rs
+++ b/datafusion/functions/src/datetime/to_timestamp.rs
@@ -19,26 +19,38 @@ use std::any::Any;
 use std::sync::Arc;
 
 use crate::datetime::common::*;
+use arrow::array::timezone::Tz;
 use arrow::array::Float64Array;
 use arrow::datatypes::DataType::*;
 use arrow::datatypes::TimeUnit::{Microsecond, Millisecond, Nanosecond, Second};
 use arrow::datatypes::{
-    ArrowTimestampType, DataType, TimeUnit, TimestampMicrosecondType,
-    TimestampMillisecondType, TimestampNanosecondType, TimestampSecondType,
+    ArrowTimestampType, DataType, TimestampMicrosecondType, TimestampMillisecondType,
+    TimestampNanosecondType, TimestampSecondType,
 };
+use datafusion_common::config::ConfigOptions;
 use datafusion_common::format::DEFAULT_CAST_OPTIONS;
 use datafusion_common::{exec_err, Result, ScalarType, ScalarValue};
 use datafusion_expr::{
-    ColumnarValue, Documentation, ScalarUDFImpl, Signature, Volatility,
+    ColumnarValue, Documentation, ScalarUDF, ScalarUDFImpl, Signature, Volatility,
 };
 use datafusion_macros::user_doc;
 
 #[user_doc(
     doc_section(label = "Time and Date Functions"),
     description = r#"
-Converts a value to a timestamp (`YYYY-MM-DDT00:00:00Z`). Supports strings, integer, unsigned integer, and double types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00') if no [Chrono formats] are provided. Integers, unsigned integers, and doubles are interpreted as seconds since the unix epoch (`1970-01-01T00:00:00Z`). Returns the corresponding timestamp.
-
-Note: `to_timestamp` returns `Timestamp(ns)`. The supported range for integer input is between `-9223372037` and `9223372036`. Supported range for string input is between `1677-09-21T00:12:44.0` and `2262-04-11T23:47:16.0`. Please use `to_timestamp_seconds` for the input outside of supported bounds.
+Converts a value to a timestamp (`YYYY-MM-DDT00:00:00.000000<TZ>`) in the session time zone. Supports strings,
+integer, unsigned integer, and double types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00')
+if no [Chrono formats] are provided. Strings that parse without a time zone are treated as if they are in the
+session time zone, or UTC if no session time zone is set.
+Integers, unsigned integers, and doubles are interpreted as seconds since the unix epoch (`1970-01-01T00:00:00Z`).
+
+Note: `to_timestamp` returns `Timestamp(ns, TimeZone)` where the time zone is the session time zone. The supported range
+for integer input is between`-9223372037` and `9223372036`. Supported range for string input is between
+`1677-09-21T00:12:44.0` and `2262-04-11T23:47:16.0`. Please use `to_timestamp_seconds`
+for the input outside of supported bounds.
+
+The session time zone can be set using the statement `SET TIMEZONE = 'desired time zone'`.
+The time zone can be a value like +00:00, 'Europe/London' etc.
 "#,
     syntax_example = "to_timestamp(expression[, ..., format_n])",
     sql_example = r#"```sql
@@ -63,17 +75,32 @@ Additional examples can be found [here](https://github.com/apache/datafusion/blo
     ),
     argument(
         name = "format_n",
-        description = "Optional [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) strings to use to parse the expression. Formats will be tried in the order they appear with the first successful one being returned. If none of the formats successfully parse the expression an error will be returned."
+        description = r#"
+Optional [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) strings to use to parse the expression.
+Formats will be tried in the order they appear with the first successful one being returned. If none of the formats successfully
+parse the expression an error will be returned. Note: parsing of named timezones (e.g. 'America/New_York') using %Z is
+only supported at the end of the string preceded by a space.
+"#
     )
 )]
 #[derive(Debug, PartialEq, Eq, Hash)]
 pub struct ToTimestampFunc {
     signature: Signature,
+    timezone: Option<Arc<str>>,
 }
 
 #[user_doc(
     doc_section(label = "Time and Date Functions"),
-    description = "Converts a value to a timestamp (`YYYY-MM-DDT00:00:00.000Z`). Supports strings, integer, and unsigned integer types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00') if no [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html)s are provided. Integers and unsigned integers are interpreted as seconds since the unix epoch (`1970-01-01T00:00:00Z`). Returns the corresponding timestamp.",
+    description = r#"
+Converts a value to a timestamp (`YYYY-MM-DDT00:00:00<TZ>`) in the session time zone. Supports strings,
+integer, unsigned integer, and double types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00')
+if no [Chrono formats] are provided. Strings that parse without a time zone are treated as if they are in the
+session time zone, or UTC if no session time zone is set.
+Integers, unsigned integers, and doubles are interpreted as seconds since the unix epoch (`1970-01-01T00:00:00Z`).
+
+The session time zone can be set using the statement `SET TIMEZONE = 'desired time zone'`.
+The time zone can be a value like +00:00, 'Europe/London' etc.
+"#,
     syntax_example = "to_timestamp_seconds(expression[, ..., format_n])",
     sql_example = r#"```sql
 > select to_timestamp_seconds('2023-01-31T09:26:56.123456789-05:00');
@@ -97,17 +124,32 @@ Additional examples can be found [here](https://github.com/apache/datafusion/blo
     ),
     argument(
         name = "format_n",
-        description = "Optional [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) strings to use to parse the expression. Formats will be tried in the order they appear with the first successful one being returned. If none of the formats successfully parse the expression an error will be returned."
+        description = r#"
+Optional [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) strings to use to parse the expression.
+Formats will be tried in the order they appear with the first successful one being returned. If none of the formats successfully
+parse the expression an error will be returned. Note: parsing of named timezones (e.g. 'America/New_York') using %Z is
+only supported at the end of the string preceded by a space.
+"#
     )
 )]
 #[derive(Debug, PartialEq, Eq, Hash)]
 pub struct ToTimestampSecondsFunc {
     signature: Signature,
+    timezone: Option<Arc<str>>,
 }
 
 #[user_doc(
     doc_section(label = "Time and Date Functions"),
-    description = "Converts a value to a timestamp (`YYYY-MM-DDT00:00:00.000Z`). Supports strings, integer, and unsigned integer types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00') if no [Chrono formats](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) are provided. Integers and unsigned integers are interpreted as milliseconds since the unix epoch (`1970-01-01T00:00:00Z`). Returns the corresponding timestamp.",
+    description = r#"
+Converts a value to a timestamp (`YYYY-MM-DDT00:00:00.000<TZ>`) in the session time zone. Supports strings,
+integer, unsigned integer, and double types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00')
+if no [Chrono formats] are provided. Strings that parse without a time zone are treated as if they are in the
+session time zone, or UTC if no session time zone is set.
+Integers, unsigned integers, and doubles are interpreted as milliseconds since the unix epoch (`1970-01-01T00:00:00Z`).
+
+The session time zone can be set using the statement `SET TIMEZONE = 'desired time zone'`.
+The time zone can be a value like +00:00, 'Europe/London' etc.
+"#,
     syntax_example = "to_timestamp_millis(expression[, ..., format_n])",
     sql_example = r#"```sql
 > select to_timestamp_millis('2023-01-31T09:26:56.123456789-05:00');
@@ -131,17 +173,32 @@ Additional examples can be found [here](https://github.com/apache/datafusion/blo
     ),
     argument(
         name = "format_n",
-        description = "Optional [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) strings to use to parse the expression. Formats will be tried in the order they appear with the first successful one being returned. If none of the formats successfully parse the expression an error will be returned."
+        description = r#"
+Optional [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) strings to use to parse the expression.
+Formats will be tried in the order they appear with the first successful one being returned. If none of the formats successfully
+parse the expression an error will be returned. Note: parsing of named timezones (e.g. 'America/New_York') using %Z is
+only supported at the end of the string preceded by a space.
+"#
     )
 )]
 #[derive(Debug, PartialEq, Eq, Hash)]
 pub struct ToTimestampMillisFunc {
     signature: Signature,
+    timezone: Option<Arc<str>>,
 }
 
 #[user_doc(
     doc_section(label = "Time and Date Functions"),
-    description = "Converts a value to a timestamp (`YYYY-MM-DDT00:00:00.000000Z`). Supports strings, integer, and unsigned integer types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00') if no [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html)s are provided. Integers and unsigned integers are interpreted as microseconds since the unix epoch (`1970-01-01T00:00:00Z`) Returns the corresponding timestamp.",
+    description = r#"
+Converts a value to a timestamp (`YYYY-MM-DDT00:00:00.000000<TZ>`) in the session time zone. Supports strings,
+integer, unsigned integer, and double types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00')
+if no [Chrono formats] are provided. Strings that parse without a time zone are treated as if they are in the
+session time zone, or UTC if no session time zone is set.
+Integers, unsigned integers, and doubles are interpreted as microseconds since the unix epoch (`1970-01-01T00:00:00Z`).
+
+The session time zone can be set using the statement `SET TIMEZONE = 'desired time zone'`.
+The time zone can be a value like +00:00, 'Europe/London' etc.
+"#,
     syntax_example = "to_timestamp_micros(expression[, ..., format_n])",
     sql_example = r#"```sql
 > select to_timestamp_micros('2023-01-31T09:26:56.123456789-05:00');
@@ -165,17 +222,31 @@ Additional examples can be found [here](https://github.com/apache/datafusion/blo
     ),
     argument(
         name = "format_n",
-        description = "Optional [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) strings to use to parse the expression. Formats will be tried in the order they appear with the first successful one being returned. If none of the formats successfully parse the expression an error will be returned."
+        description = r#"
+Optional [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) strings to use to parse the expression.
+Formats will be tried in the order they appear with the first successful one being returned. If none of the formats successfully
+parse the expression an error will be returned. Note: parsing of named timezones (e.g. 'America/New_York') using %Z is
+only supported at the end of the string preceded by a space.
+"#
     )
 )]
 #[derive(Debug, PartialEq, Eq, Hash)]
 pub struct ToTimestampMicrosFunc {
     signature: Signature,
+    timezone: Option<Arc<str>>,
 }
 
 #[user_doc(
     doc_section(label = "Time and Date Functions"),
-    description = "Converts a value to a timestamp (`YYYY-MM-DDT00:00:00.000000000Z`). Supports strings, integer, and unsigned integer types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00') if no [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html)s are provided. Integers and unsigned integers are interpreted as nanoseconds since the unix epoch (`1970-01-01T00:00:00Z`). Returns the corresponding timestamp.",
+    description = r#"
+Converts a value to a timestamp (`YYYY-MM-DDT00:00:00.000000000Z<TZ>`) in the session time zone. Supports strings,
+integer, unsigned integer, and double types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00')
+if no [Chrono formats] are provided. Strings that parse without a time zone are treated as if they are in the
+session time zone. Integers, unsigned integers, and doubles are interpreted as nanoseconds since the unix epoch (`1970-01-01T00:00:00Z`).
+
+The session time zone can be set using the statement `SET TIMEZONE = 'desired time zone'`.
+The time zone can be a value like +00:00, 'Europe/London' etc.
+"#,
     syntax_example = "to_timestamp_nanos(expression[, ..., format_n])",
     sql_example = r#"```sql
 > select to_timestamp_nanos('2023-01-31T09:26:56.123456789-05:00');
@@ -199,80 +270,161 @@ Additional examples can be found [here](https://github.com/apache/datafusion/blo
     ),
     argument(
         name = "format_n",
-        description = "Optional [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) strings to use to parse the expression. Formats will be tried in the order they appear with the first successful one being returned. If none of the formats successfully parse the expression an error will be returned."
+        description = r#"
+Optional [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) strings to use to parse the expression.
+Formats will be tried in the order they appear with the first successful one being returned. If none of the formats successfully
+parse the expression an error will be returned. Note: parsing of named timezones (e.g. 'America/New_York') using %Z is
+only supported at the end of the string preceded by a space.
+"#
     )
 )]
 #[derive(Debug, PartialEq, Eq, Hash)]
 pub struct ToTimestampNanosFunc {
     signature: Signature,
+    timezone: Option<Arc<str>>,
 }
 
 impl Default for ToTimestampFunc {
     fn default() -> Self {
-        Self::new()
+        Self::new_with_config(&ConfigOptions::default())
     }
 }
 
 impl ToTimestampFunc {
+    #[deprecated(since = "52.0.0", note = "use `new_with_config` instead")]
+    /// Deprecated constructor retained for backwards compatibility.
+    ///
+    /// Prefer [`ToTimestampFunc::new_with_config`] which allows specifying the
+    /// timezone via [`ConfigOptions`]. This helper now mirrors the
+    /// canonical default offset (None) provided by `ConfigOptions::default()`.
     pub fn new() -> Self {
+        Self::new_with_config(&ConfigOptions::default())
+    }
+
+    pub fn new_with_config(config: &ConfigOptions) -> Self {
         Self {
             signature: Signature::variadic_any(Volatility::Immutable),
+            timezone: config
+                .execution
+                .time_zone
+                .as_ref()
+                .map(|tz| Arc::from(tz.as_str())),
         }
     }
 }
 
 impl Default for ToTimestampSecondsFunc {
     fn default() -> Self {
-        Self::new()
+        Self::new_with_config(&ConfigOptions::default())
     }
 }
 
 impl ToTimestampSecondsFunc {
+    #[deprecated(since = "52.0.0", note = "use `new_with_config` instead")]
+    /// Deprecated constructor retained for backwards compatibility.
+    ///
+    /// Prefer [`ToTimestampSecondsFunc::new_with_config`] which allows specifying the
+    /// timezone via [`ConfigOptions`]. This helper now mirrors the
+    /// canonical default offset (None) provided by `ConfigOptions::default()`.
     pub fn new() -> Self {
+        Self::new_with_config(&ConfigOptions::default())
+    }
+
+    pub fn new_with_config(config: &ConfigOptions) -> Self {
         Self {
             signature: Signature::variadic_any(Volatility::Immutable),
+            timezone: config
+                .execution
+                .time_zone
+                .as_ref()
+                .map(|tz| Arc::from(tz.as_str())),
         }
     }
 }
 
 impl Default for ToTimestampMillisFunc {
     fn default() -> Self {
-        Self::new()
+        Self::new_with_config(&ConfigOptions::default())
     }
 }
 
 impl ToTimestampMillisFunc {
+    #[deprecated(since = "52.0.0", note = "use `new_with_config` instead")]
+    /// Deprecated constructor retained for backwards compatibility.
+    ///
+    /// Prefer [`ToTimestampMillisFunc::new_with_config`] which allows specifying the
+    /// timezone via [`ConfigOptions`]. This helper now mirrors the
+    /// canonical default offset (None) provided by `ConfigOptions::default()`.
     pub fn new() -> Self {
+        Self::new_with_config(&ConfigOptions::default())
+    }
+
+    pub fn new_with_config(config: &ConfigOptions) -> Self {
         Self {
             signature: Signature::variadic_any(Volatility::Immutable),
+            timezone: config
+                .execution
+                .time_zone
+                .as_ref()
+                .map(|tz| Arc::from(tz.as_str())),
         }
     }
 }
 
 impl Default for ToTimestampMicrosFunc {
     fn default() -> Self {
-        Self::new()
+        Self::new_with_config(&ConfigOptions::default())
     }
 }
 
 impl ToTimestampMicrosFunc {
+    #[deprecated(since = "52.0.0", note = "use `new_with_config` instead")]
+    /// Deprecated constructor retained for backwards compatibility.
+    ///
+    /// Prefer [`ToTimestampMicrosFunc::new_with_config`] which allows specifying the
+    /// timezone via [`ConfigOptions`]. This helper now mirrors the
+    /// canonical default offset (None) provided by `ConfigOptions::default()`.
     pub fn new() -> Self {
+        Self::new_with_config(&ConfigOptions::default())
+    }
+
+    pub fn new_with_config(config: &ConfigOptions) -> Self {
         Self {
             signature: Signature::variadic_any(Volatility::Immutable),
+            timezone: config
+                .execution
+                .time_zone
+                .as_ref()
+                .map(|tz| Arc::from(tz.as_str())),
         }
     }
 }
 
 impl Default for ToTimestampNanosFunc {
     fn default() -> Self {
-        Self::new()
+        Self::new_with_config(&ConfigOptions::default())
     }
 }
 
 impl ToTimestampNanosFunc {
+    #[deprecated(since = "52.0.0", note = "use `new_with_config` instead")]
+    /// Deprecated constructor retained for backwards compatibility.
+    ///
+    /// Prefer [`ToTimestampNanosFunc::new_with_config`] which allows specifying the
+    /// timezone via [`ConfigOptions`]. This helper now mirrors the
+    /// canonical default offset (None) provided by `ConfigOptions::default()`.
     pub fn new() -> Self {
+        Self::new_with_config(&ConfigOptions::default())
+    }
+
+    pub fn new_with_config(config: &ConfigOptions) -> Self {
         Self {
             signature: Signature::variadic_any(Volatility::Immutable),
+            timezone: config
+                .execution
+                .time_zone
+                .as_ref()
+                .map(|tz| Arc::from(tz.as_str())),
         }
     }
 }
@@ -296,15 +448,20 @@ impl ScalarUDFImpl for ToTimestampFunc {
         &self.signature
     }
 
-    fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
-        Ok(return_type_for(&arg_types[0], Nanosecond))
+    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+        Ok(Timestamp(Nanosecond, self.timezone.clone()))
+    }
+
+    fn with_updated_config(&self, config: &ConfigOptions) -> Option<ScalarUDF> {
+        Some(Self::new_with_config(config).into())
     }
 
     fn invoke_with_args(
         &self,
         args: datafusion_expr::ScalarFunctionArgs,
     ) -> Result<ColumnarValue> {
-        let args = args.args;
+        let datafusion_expr::ScalarFunctionArgs { args, .. } = args;
+
         if args.is_empty() {
             return exec_err!(
                 "to_timestamp function requires 1 or more arguments, got {}",
@@ -317,13 +474,13 @@ impl ScalarUDFImpl for ToTimestampFunc {
             validate_data_types(&args, "to_timestamp")?;
         }
 
+        let tz = self.timezone.clone();
+
         match args[0].data_type() {
             Int32 | Int64 => args[0]
                 .cast_to(&Timestamp(Second, None), None)?
-                .cast_to(&Timestamp(Nanosecond, None), None),
-            Null | Timestamp(_, None) => {
-                args[0].cast_to(&Timestamp(Nanosecond, None), None)
-            }
+                .cast_to(&Timestamp(Nanosecond, tz), None),
+            Null | Timestamp(_, _) => args[0].cast_to(&Timestamp(Nanosecond, tz), None),
             Float64 => {
                 let rescaled = arrow::compute::kernels::numeric::mul(
                     &args[0].to_array(1)?,
@@ -333,15 +490,12 @@ impl ScalarUDFImpl for ToTimestampFunc {
                 )?;
                 Ok(ColumnarValue::Array(arrow::compute::cast_with_options(
                     &rescaled,
-                    &Timestamp(Nanosecond, None),
+                    &Timestamp(Nanosecond, tz),
                     &DEFAULT_CAST_OPTIONS,
                 )?))
             }
-            Timestamp(_, Some(tz)) => {
-                args[0].cast_to(&Timestamp(Nanosecond, Some(tz)), None)
-            }
             Utf8View | LargeUtf8 | Utf8 => {
-                to_timestamp_impl::<TimestampNanosecondType>(&args, "to_timestamp")
+                to_timestamp_impl::<TimestampNanosecondType>(&args, "to_timestamp", &tz)
             }
             Decimal128(_, _) => {
                 match &args[0] {
@@ -354,14 +508,12 @@ impl ScalarUDFImpl for ToTimestampFunc {
                         let scale_factor = 10_i128.pow(*scale as u32);
                         let seconds = value / scale_factor;
                         let fraction = value % scale_factor;
-
                         let nanos = (fraction * 1_000_000_000) / scale_factor;
-
                         let timestamp_nanos = seconds * 1_000_000_000 + nanos;
 
                         Ok(ColumnarValue::Scalar(ScalarValue::TimestampNanosecond(
                             Some(timestamp_nanos as i64),
-                            None,
+                            tz,
                         )))
                     }
                     _ => exec_err!("Invalid decimal value"),
@@ -372,6 +524,7 @@ impl ScalarUDFImpl for ToTimestampFunc {
             }
         }
     }
+
     fn documentation(&self) -> Option<&Documentation> {
         self.doc()
     }
@@ -390,15 +543,20 @@ impl ScalarUDFImpl for ToTimestampSecondsFunc {
         &self.signature
     }
 
-    fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
-        Ok(return_type_for(&arg_types[0], Second))
+    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+        Ok(Timestamp(Second, self.timezone.clone()))
+    }
+
+    fn with_updated_config(&self, config: &ConfigOptions) -> Option<ScalarUDF> {
+        Some(Self::new_with_config(config).into())
     }
 
     fn invoke_with_args(
         &self,
         args: datafusion_expr::ScalarFunctionArgs,
     ) -> Result<ColumnarValue> {
-        let args = args.args;
+        let datafusion_expr::ScalarFunctionArgs { args, .. } = args;
+
         if args.is_empty() {
             return exec_err!(
                 "to_timestamp_seconds function requires 1 or more arguments, got {}",
@@ -411,14 +569,17 @@ impl ScalarUDFImpl for ToTimestampSecondsFunc {
             validate_data_types(&args, "to_timestamp")?;
         }
 
+        let tz = self.timezone.clone();
+
         match args[0].data_type() {
-            Null | Int32 | Int64 | Timestamp(_, None) | Decimal128(_, _) => {
-                args[0].cast_to(&Timestamp(Second, None), None)
-            }
-            Timestamp(_, Some(tz)) => args[0].cast_to(&Timestamp(Second, Some(tz)), None),
-            Utf8View | LargeUtf8 | Utf8 => {
-                to_timestamp_impl::<TimestampSecondType>(&args, "to_timestamp_seconds")
+            Null | Int32 | Int64 | Timestamp(_, _) | Decimal128(_, _) => {
+                args[0].cast_to(&Timestamp(Second, tz), None)
             }
+            Utf8View | LargeUtf8 | Utf8 => to_timestamp_impl::<TimestampSecondType>(
+                &args,
+                "to_timestamp_seconds",
+                &self.timezone,
+            ),
             other => {
                 exec_err!(
                     "Unsupported data type {} for function to_timestamp_seconds",
@@ -427,6 +588,7 @@ impl ScalarUDFImpl for ToTimestampSecondsFunc {
             }
         }
     }
+
     fn documentation(&self) -> Option<&Documentation> {
         self.doc()
     }
@@ -445,15 +607,20 @@ impl ScalarUDFImpl for ToTimestampMillisFunc {
         &self.signature
     }
 
-    fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
-        Ok(return_type_for(&arg_types[0], Millisecond))
+    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+        Ok(Timestamp(Millisecond, self.timezone.clone()))
+    }
+
+    fn with_updated_config(&self, config: &ConfigOptions) -> Option<ScalarUDF> {
+        Some(Self::new_with_config(config).into())
     }
 
     fn invoke_with_args(
         &self,
         args: datafusion_expr::ScalarFunctionArgs,
     ) -> Result<ColumnarValue> {
-        let args = args.args;
+        let datafusion_expr::ScalarFunctionArgs { args, .. } = args;
+
         if args.is_empty() {
             return exec_err!(
                 "to_timestamp_millis function requires 1 or more arguments, got {}",
@@ -467,15 +634,13 @@ impl ScalarUDFImpl for ToTimestampMillisFunc {
         }
 
         match args[0].data_type() {
-            Null | Int32 | Int64 | Timestamp(_, None) => {
-                args[0].cast_to(&Timestamp(Millisecond, None), None)
-            }
-            Timestamp(_, Some(tz)) => {
-                args[0].cast_to(&Timestamp(Millisecond, Some(tz)), None)
+            Null | Int32 | Int64 | Timestamp(_, _) => {
+                args[0].cast_to(&Timestamp(Millisecond, self.timezone.clone()), None)
             }
             Utf8View | LargeUtf8 | Utf8 => to_timestamp_impl::<TimestampMillisecondType>(
                 &args,
                 "to_timestamp_millis",
+                &self.timezone,
             ),
             other => {
                 exec_err!(
@@ -485,6 +650,7 @@ impl ScalarUDFImpl for ToTimestampMillisFunc {
             }
         }
     }
+
     fn documentation(&self) -> Option<&Documentation> {
         self.doc()
     }
@@ -503,15 +669,20 @@ impl ScalarUDFImpl for ToTimestampMicrosFunc {
         &self.signature
     }
 
-    fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
-        Ok(return_type_for(&arg_types[0], Microsecond))
+    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+        Ok(Timestamp(Microsecond, self.timezone.clone()))
+    }
+
+    fn with_updated_config(&self, config: &ConfigOptions) -> Option<ScalarUDF> {
+        Some(Self::new_with_config(config).into())
     }
 
     fn invoke_with_args(
         &self,
         args: datafusion_expr::ScalarFunctionArgs,
     ) -> Result<ColumnarValue> {
-        let args = args.args;
+        let datafusion_expr::ScalarFunctionArgs { args, .. } = args;
+
         if args.is_empty() {
             return exec_err!(
                 "to_timestamp_micros function requires 1 or more arguments, got {}",
@@ -525,15 +696,13 @@ impl ScalarUDFImpl for ToTimestampMicrosFunc {
         }
 
         match args[0].data_type() {
-            Null | Int32 | Int64 | Timestamp(_, None) => {
-                args[0].cast_to(&Timestamp(Microsecond, None), None)
-            }
-            Timestamp(_, Some(tz)) => {
-                args[0].cast_to(&Timestamp(Microsecond, Some(tz)), None)
+            Null | Int32 | Int64 | Timestamp(_, _) => {
+                args[0].cast_to(&Timestamp(Microsecond, self.timezone.clone()), None)
             }
             Utf8View | LargeUtf8 | Utf8 => to_timestamp_impl::<TimestampMicrosecondType>(
                 &args,
                 "to_timestamp_micros",
+                &self.timezone,
             ),
             other => {
                 exec_err!(
@@ -543,6 +712,7 @@ impl ScalarUDFImpl for ToTimestampMicrosFunc {
             }
         }
     }
+
     fn documentation(&self) -> Option<&Documentation> {
         self.doc()
     }
@@ -561,15 +731,20 @@ impl ScalarUDFImpl for ToTimestampNanosFunc {
         &self.signature
     }
 
-    fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
-        Ok(return_type_for(&arg_types[0], Nanosecond))
+    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+        Ok(Timestamp(Nanosecond, self.timezone.clone()))
+    }
+
+    fn with_updated_config(&self, config: &ConfigOptions) -> Option<ScalarUDF> {
+        Some(Self::new_with_config(config).into())
     }
 
     fn invoke_with_args(
         &self,
         args: datafusion_expr::ScalarFunctionArgs,
     ) -> Result<ColumnarValue> {
-        let args = args.args;
+        let datafusion_expr::ScalarFunctionArgs { args, .. } = args;
+
         if args.is_empty() {
             return exec_err!(
                 "to_timestamp_nanos function requires 1 or more arguments, got {}",
@@ -583,15 +758,14 @@ impl ScalarUDFImpl for ToTimestampNanosFunc {
         }
 
         match args[0].data_type() {
-            Null | Int32 | Int64 | Timestamp(_, None) => {
-                args[0].cast_to(&Timestamp(Nanosecond, None), None)
-            }
-            Timestamp(_, Some(tz)) => {
-                args[0].cast_to(&Timestamp(Nanosecond, Some(tz)), None)
-            }
-            Utf8View | LargeUtf8 | Utf8 => {
-                to_timestamp_impl::<TimestampNanosecondType>(&args, "to_timestamp_nanos")
+            Null | Int32 | Int64 | Timestamp(_, _) => {
+                args[0].cast_to(&Timestamp(Nanosecond, self.timezone.clone()), None)
             }
+            Utf8View | LargeUtf8 | Utf8 => to_timestamp_impl::<TimestampNanosecondType>(
+                &args,
+                "to_timestamp_nanos",
+                &self.timezone,
+            ),
             other => {
                 exec_err!(
                     "Unsupported data type {} for function to_timestamp_nanos",
@@ -600,23 +774,16 @@ impl ScalarUDFImpl for ToTimestampNanosFunc {
             }
         }
     }
+
     fn documentation(&self) -> Option<&Documentation> {
         self.doc()
     }
 }
 
-/// Returns the return type for the to_timestamp_* function, preserving
-/// the timezone if it exists.
-fn return_type_for(arg: &DataType, unit: TimeUnit) -> DataType {
-    match arg {
-        Timestamp(_, Some(tz)) => Timestamp(unit, Some(Arc::clone(tz))),
-        _ => Timestamp(unit, None),
-    }
-}
-
 fn to_timestamp_impl<T: ArrowTimestampType + ScalarType<i64>>(
     args: &[ColumnarValue],
     name: &str,
+    timezone: &Option<Arc<str>>,
 ) -> Result<ColumnarValue> {
     let factor = match T::UNIT {
         Second => 1_000_000_000,
@@ -625,17 +792,26 @@ fn to_timestamp_impl<T: ArrowTimestampType + ScalarType<i64>>(
         Nanosecond => 1,
     };
 
+    let tz = match timezone.clone() {
+        Some(tz) => Some(tz.parse::<Tz>()?),
+        None => None,
+    };
+
     match args.len() {
-        1 => handle::<T, _, T>(
+        1 => handle::<T, _, i64>(
             args,
-            |s| string_to_timestamp_nanos_shim(s).map(|n| n / factor),
+            move |s| string_to_timestamp_nanos_with_timezone(&tz, s).map(|n| n / factor),
             name,
+            &ScalarDataType::new(Timestamp(T::UNIT, timezone.clone())),
         ),
-        n if n >= 2 => handle_multiple::<T, _, T, _>(
+        n if n >= 2 => handle_multiple::<T, _, _, i64>(
             args,
-            string_to_timestamp_nanos_formatted,
+            move |s, format| {
+                string_to_timestamp_nanos_formatted_with_timezone(&tz, s, format)
+            },
             |n| n / factor,
             name,
+            &ScalarDataType::new(Timestamp(T::UNIT, timezone.clone())),
         ),
         _ => exec_err!("Unsupported 0 argument count for function {name}"),
     }
@@ -652,35 +828,110 @@ mod tests {
     };
     use arrow::array::{ArrayRef, Int64Array, StringBuilder};
     use arrow::datatypes::{Field, TimeUnit};
-    use chrono::Utc;
+    use chrono::{DateTime, FixedOffset, Utc};
     use datafusion_common::config::ConfigOptions;
     use datafusion_common::{assert_contains, DataFusionError, ScalarValue};
-    use datafusion_expr::ScalarFunctionImplementation;
+    use datafusion_expr::{ScalarFunctionArgs, ScalarFunctionImplementation};
 
     use super::*;
 
     fn to_timestamp(args: &[ColumnarValue]) -> Result<ColumnarValue> {
-        to_timestamp_impl::<TimestampNanosecondType>(args, "to_timestamp")
+        let timezone: Option<Arc<str>> = Some("UTC".into());
+        to_timestamp_impl::<TimestampNanosecondType>(args, "to_timestamp", &timezone)
     }
 
     /// to_timestamp_millis SQL function
     fn to_timestamp_millis(args: &[ColumnarValue]) -> Result<ColumnarValue> {
-        to_timestamp_impl::<TimestampMillisecondType>(args, "to_timestamp_millis")
+        let timezone: Option<Arc<str>> = Some("UTC".into());
+        to_timestamp_impl::<TimestampMillisecondType>(
+            args,
+            "to_timestamp_millis",
+            &timezone,
+        )
     }
 
     /// to_timestamp_micros SQL function
     fn to_timestamp_micros(args: &[ColumnarValue]) -> Result<ColumnarValue> {
-        to_timestamp_impl::<TimestampMicrosecondType>(args, "to_timestamp_micros")
+        let timezone: Option<Arc<str>> = Some("UTC".into());
+        to_timestamp_impl::<TimestampMicrosecondType>(
+            args,
+            "to_timestamp_micros",
+            &timezone,
+        )
     }
 
     /// to_timestamp_nanos SQL function
     fn to_timestamp_nanos(args: &[ColumnarValue]) -> Result<ColumnarValue> {
-        to_timestamp_impl::<TimestampNanosecondType>(args, "to_timestamp_nanos")
+        let timezone: Option<Arc<str>> = Some("UTC".into());
+        to_timestamp_impl::<TimestampNanosecondType>(
+            args,
+            "to_timestamp_nanos",
+            &timezone,
+        )
     }
 
     /// to_timestamp_seconds SQL function
     fn to_timestamp_seconds(args: &[ColumnarValue]) -> Result<ColumnarValue> {
-        to_timestamp_impl::<TimestampSecondType>(args, "to_timestamp_seconds")
+        let timezone: Option<Arc<str>> = Some("UTC".into());
+        to_timestamp_impl::<TimestampSecondType>(args, "to_timestamp_seconds", &timezone)
+    }
+
+    fn udfs_and_timeunit() -> Vec<(Box<dyn ScalarUDFImpl>, TimeUnit)> {
+        let udfs: Vec<(Box<dyn ScalarUDFImpl>, TimeUnit)> = vec![
+            (
+                Box::new(ToTimestampFunc::new_with_config(&ConfigOptions::default())),
+                Nanosecond,
+            ),
+            (
+                Box::new(ToTimestampSecondsFunc::new_with_config(
+                    &ConfigOptions::default(),
+                )),
+                Second,
+            ),
+            (
+                Box::new(ToTimestampMillisFunc::new_with_config(
+                    &ConfigOptions::default(),
+                )),
+                Millisecond,
+            ),
+            (
+                Box::new(ToTimestampMicrosFunc::new_with_config(
+                    &ConfigOptions::default(),
+                )),
+                Microsecond,
+            ),
+            (
+                Box::new(ToTimestampNanosFunc::new_with_config(
+                    &ConfigOptions::default(),
+                )),
+                Nanosecond,
+            ),
+        ];
+        udfs
+    }
+
+    fn validate_expected_error(
+        options: &mut ConfigOptions,
+        args: ScalarFunctionArgs,
+        expected_err: &str,
+    ) {
+        let udfs = udfs_and_timeunit();
+
+        for (udf, _) in udfs {
+            match udf
+                .with_updated_config(options)
+                .unwrap()
+                .invoke_with_args(args.clone())
+            {
+                Ok(_) => panic!("Expected error but got success"),
+                Err(e) => {
+                    assert!(
+                        e.to_string().contains(expected_err),
+                        "Can not find expected error '{expected_err}'. Actual error '{e}'"
+                    );
+                }
+            }
+        }
     }
 
     #[test]
@@ -751,6 +1002,368 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn to_timestamp_respects_execution_timezone() -> Result<()> {
+        let udfs = udfs_and_timeunit();
+
+        let mut options = ConfigOptions::default();
+        options.execution.time_zone = Some("-05:00".to_string());
+
+        let time_zone: Option<Arc<str>> = options
+            .execution
+            .time_zone
+            .as_ref()
+            .map(|tz| Arc::from(tz.as_str()));
+
+        for (udf, time_unit) in udfs {
+            let field = Field::new("arg", Utf8, true).into();
+
+            let args = ScalarFunctionArgs {
+                args: vec![ColumnarValue::Scalar(ScalarValue::Utf8(Some(
+                    "2020-09-08T13:42:29".to_string(),
+                )))],
+                arg_fields: vec![field],
+                number_rows: 1,
+                return_field: Field::new(
+                    "f",
+                    Timestamp(time_unit, Some("-05:00".into())),
+                    true,
+                )
+                .into(),
+                config_options: Arc::new(options.clone()),
+            };
+
+            let result = udf
+                .with_updated_config(&options.clone())
+                .unwrap()
+                .invoke_with_args(args)?;
+            let result = match time_unit {
+                Second => {
+                    let ColumnarValue::Scalar(ScalarValue::TimestampSecond(
+                        Some(value),
+                        tz,
+                    )) = result
+                    else {
+                        panic!("expected scalar timestamp");
+                    };
+
+                    assert_eq!(tz, time_zone);
+
+                    value
+                }
+                Millisecond => {
+                    let ColumnarValue::Scalar(ScalarValue::TimestampMillisecond(
+                        Some(value),
+                        tz,
+                    )) = result
+                    else {
+                        panic!("expected scalar timestamp");
+                    };
+
+                    assert_eq!(tz, time_zone);
+
+                    value
+                }
+                Microsecond => {
+                    let ColumnarValue::Scalar(ScalarValue::TimestampMicrosecond(
+                        Some(value),
+                        tz,
+                    )) = result
+                    else {
+                        panic!("expected scalar timestamp");
+                    };
+
+                    assert_eq!(tz, time_zone);
+
+                    value
+                }
+                Nanosecond => {
+                    let ColumnarValue::Scalar(ScalarValue::TimestampNanosecond(
+                        Some(value),
+                        tz,
+                    )) = result
+                    else {
+                        panic!("expected scalar timestamp");
+                    };
+
+                    assert_eq!(tz, time_zone);
+
+                    value
+                }
+            };
+
+            let scale = match time_unit {
+                Second => 1_000_000_000,
+                Millisecond => 1_000_000,
+                Microsecond => 1_000,
+                Nanosecond => 1,
+            };
+
+            let offset = FixedOffset::west_opt(5 * 3600).unwrap();
+            let result = Some(
+                DateTime::<Utc>::from_timestamp_nanos(result * scale)
+                    .with_timezone(&offset)
+                    .to_string(),
+            );
+
+            assert_eq!(result, Some("2020-09-08 13:42:29 -05:00".to_string()));
+        }
+
+        Ok(())
+    }
+
+    #[test]
+    fn to_timestamp_formats_respects_execution_timezone() -> Result<()> {
+        let udfs = udfs_and_timeunit();
+
+        let mut options = ConfigOptions::default();
+        options.execution.time_zone = Some("-05:00".to_string());
+
+        let time_zone: Option<Arc<str>> = options
+            .execution
+            .time_zone
+            .as_ref()
+            .map(|tz| Arc::from(tz.as_str()));
+
+        let expr_field = Field::new("arg", Utf8, true).into();
+        let format_field: Arc<Field> = Field::new("fmt", Utf8, true).into();
+
+        for (udf, time_unit) in udfs {
+            for (value, format, expected_str) in [
+                (
+                    "2020-09-08 09:42:29 -05:00",
+                    "%Y-%m-%d %H:%M:%S %z",
+                    Some("2020-09-08 09:42:29 -05:00"),
+                ),
+                (
+                    "2020-09-08T13:42:29Z",
+                    "%+",
+                    Some("2020-09-08 08:42:29 -05:00"),
+                ),
+                (
+                    "2020-09-08 13:42:29 UTC",
+                    "%Y-%m-%d %H:%M:%S %Z",
+                    Some("2020-09-08 08:42:29 -05:00"),
+                ),
+                (
+                    "+0000 2024-01-01 12:00:00",
+                    "%z %Y-%m-%d %H:%M:%S",
+                    Some("2024-01-01 07:00:00 -05:00"),
+                ),
+                (
+                    "20200908134229+0100",
+                    "%Y%m%d%H%M%S%z",
+                    Some("2020-09-08 07:42:29 -05:00"),
+                ),
+                (
+                    "2020-09-08+0230 13:42",
+                    "%Y-%m-%d%z %H:%M",
+                    Some("2020-09-08 06:12:00 -05:00"),
+                ),
+            ] {
+                let args = ScalarFunctionArgs {
+                    args: vec![
+                        ColumnarValue::Scalar(ScalarValue::Utf8(Some(value.to_string()))),
+                        ColumnarValue::Scalar(ScalarValue::Utf8(Some(
+                            format.to_string(),
+                        ))),
+                    ],
+                    arg_fields: vec![Arc::clone(&expr_field), Arc::clone(&format_field)],
+                    number_rows: 1,
+                    return_field: Field::new(
+                        "f",
+                        Timestamp(time_unit, Some("-05:00".into())),
+                        true,
+                    )
+                    .into(),
+                    config_options: Arc::new(options.clone()),
+                };
+                let result = udf
+                    .with_updated_config(&options.clone())
+                    .unwrap()
+                    .invoke_with_args(args)?;
+                let result = match time_unit {
+                    Second => {
+                        let ColumnarValue::Scalar(ScalarValue::TimestampSecond(
+                            Some(value),
+                            tz,
+                        )) = result
+                        else {
+                            panic!("expected scalar timestamp");
+                        };
+
+                        assert_eq!(tz, time_zone);
+
+                        value
+                    }
+                    Millisecond => {
+                        let ColumnarValue::Scalar(ScalarValue::TimestampMillisecond(
+                            Some(value),
+                            tz,
+                        )) = result
+                        else {
+                            panic!("expected scalar timestamp");
+                        };
+
+                        assert_eq!(tz, time_zone);
+
+                        value
+                    }
+                    Microsecond => {
+                        let ColumnarValue::Scalar(ScalarValue::TimestampMicrosecond(
+                            Some(value),
+                            tz,
+                        )) = result
+                        else {
+                            panic!("expected scalar timestamp");
+                        };
+
+                        assert_eq!(tz, time_zone);
+
+                        value
+                    }
+                    Nanosecond => {
+                        let ColumnarValue::Scalar(ScalarValue::TimestampNanosecond(
+                            Some(value),
+                            tz,
+                        )) = result
+                        else {
+                            panic!("expected scalar timestamp");
+                        };
+
+                        assert_eq!(tz, time_zone);
+
+                        value
+                    }
+                };
+
+                let scale = match time_unit {
+                    Second => 1_000_000_000,
+                    Millisecond => 1_000_000,
+                    Microsecond => 1_000,
+                    Nanosecond => 1,
+                };
+                let offset = FixedOffset::west_opt(5 * 3600).unwrap();
+                let result = Some(
+                    DateTime::<Utc>::from_timestamp_nanos(result * scale)
+                        .with_timezone(&offset)
+                        .to_string(),
+                );
+
+                assert_eq!(result, expected_str.map(|s| s.to_string()));
+            }
+        }
+
+        Ok(())
+    }
+
+    #[test]
+    fn to_timestamp_invalid_execution_timezone_behavior() -> Result<()> {
+        let field: Arc<Field> = Field::new("arg", Utf8, true).into();
+        let return_field: Arc<Field> =
+            Field::new("f", Timestamp(Nanosecond, None), true).into();
+
+        let mut options = ConfigOptions::default();
+        options.execution.time_zone = Some("Invalid/Timezone".to_string());
+
+        let args = ScalarFunctionArgs {
+            args: vec![ColumnarValue::Scalar(ScalarValue::Utf8(Some(
+                "2020-09-08T13:42:29Z".to_string(),
+            )))],
+            arg_fields: vec![Arc::clone(&field)],
+            number_rows: 1,
+            return_field: Arc::clone(&return_field),
+            config_options: Arc::new(options.clone()),
+        };
+
+        let expected_err =
+            "Invalid timezone \"Invalid/Timezone\": failed to parse timezone";
+
+        validate_expected_error(&mut options, args, expected_err);
+
+        Ok(())
+    }
+
+    #[test]
+    fn to_timestamp_formats_invalid_execution_timezone_behavior() -> Result<()> {
+        let expr_field: Arc<Field> = Field::new("arg", Utf8, true).into();
+        let format_field: Arc<Field> = Field::new("fmt", Utf8, true).into();
+        let return_field: Arc<Field> =
+            Field::new("f", Timestamp(Nanosecond, None), true).into();
+
+        let mut options = ConfigOptions::default();
+        options.execution.time_zone = Some("Invalid/Timezone".to_string());
+
+        let expected_err =
+            "Invalid timezone \"Invalid/Timezone\": failed to parse timezone";
+
+        let make_args = |value: &str, format: &str| ScalarFunctionArgs {
+            args: vec![
+                ColumnarValue::Scalar(ScalarValue::Utf8(Some(value.to_string()))),
+                ColumnarValue::Scalar(ScalarValue::Utf8(Some(format.to_string()))),
+            ],
+            arg_fields: vec![Arc::clone(&expr_field), Arc::clone(&format_field)],
+            number_rows: 1,
+            return_field: Arc::clone(&return_field),
+            config_options: Arc::new(options.clone()),
+        };
+
+        for (value, format, _expected_str) in [
+            (
+                "2020-09-08 09:42:29 -05:00",
+                "%Y-%m-%d %H:%M:%S %z",
+                Some("2020-09-08 09:42:29 -05:00"),
+            ),
+            (
+                "2020-09-08T13:42:29Z",
+                "%+",
+                Some("2020-09-08 08:42:29 -05:00"),
+            ),
+            (
+                "2020-09-08 13:42:29 +0000",
+                "%Y-%m-%d %H:%M:%S %z",
+                Some("2020-09-08 08:42:29 -05:00"),
+            ),
+            (
+                "+0000 2024-01-01 12:00:00",
+                "%z %Y-%m-%d %H:%M:%S",
+                Some("2024-01-01 07:00:00 -05:00"),
+            ),
+            (
+                "20200908134229+0100",
+                "%Y%m%d%H%M%S%z",
+                Some("2020-09-08 07:42:29 -05:00"),
+            ),
+            (
+                "2020-09-08+0230 13:42",
+                "%Y-%m-%d%z %H:%M",
+                Some("2020-09-08 06:12:00 -05:00"),
+            ),
+        ] {
+            let args = make_args(value, format);
+            validate_expected_error(&mut options.clone(), args, expected_err);
+        }
+
+        let args = ScalarFunctionArgs {
+            args: vec![
+                ColumnarValue::Scalar(ScalarValue::Utf8(Some(
+                    "2020-09-08T13:42:29".to_string(),
+                ))),
+                ColumnarValue::Scalar(ScalarValue::Utf8(Some(
+                    "%Y-%m-%dT%H:%M:%S".to_string(),
+                ))),
+            ],
+            arg_fields: vec![Arc::clone(&expr_field), Arc::clone(&format_field)],
+            number_rows: 1,
+            return_field: Arc::clone(&return_field),
+            config_options: Arc::new(options.clone()),
+        };
+
+        validate_expected_error(&mut options.clone(), args, expected_err);
+
+        Ok(())
+    }
+
     #[test]
     fn to_timestamp_invalid_input_type() -> Result<()> {
         // pass the wrong type of input array to to_timestamp and test
@@ -923,7 +1536,11 @@ mod tests {
     }
 
     fn parse_timestamp_formatted(s: &str, format: &str) -> Result<i64, DataFusionError> {
-        let result = string_to_timestamp_nanos_formatted(s, format);
+        let result = string_to_timestamp_nanos_formatted_with_timezone(
+            &Some("UTC".parse()?),
+            s,
+            format,
+        );
         if let Err(e) = &result {
             eprintln!("Error parsing timestamp '{s}' using format '{format}': {e:?}");
         }
@@ -987,13 +1604,21 @@ mod tests {
     }
 
     #[test]
-    fn test_tz() {
+    fn test_no_tz() {
         let udfs: Vec<Box<dyn ScalarUDFImpl>> = vec![
-            Box::new(ToTimestampFunc::new()),
-            Box::new(ToTimestampSecondsFunc::new()),
-            Box::new(ToTimestampMillisFunc::new()),
-            Box::new(ToTimestampNanosFunc::new()),
-            Box::new(ToTimestampSecondsFunc::new()),
+            Box::new(ToTimestampFunc::new_with_config(&ConfigOptions::default())),
+            Box::new(ToTimestampSecondsFunc::new_with_config(
+                &ConfigOptions::default(),
+            )),
+            Box::new(ToTimestampMillisFunc::new_with_config(
+                &ConfigOptions::default(),
+            )),
+            Box::new(ToTimestampNanosFunc::new_with_config(
+                &ConfigOptions::default(),
+            )),
+            Box::new(ToTimestampSecondsFunc::new_with_config(
+                &ConfigOptions::default(),
+            )),
         ];
 
         let mut nanos_builder = TimestampNanosecondArray::builder(2);
@@ -1026,8 +1651,8 @@ mod tests {
             for array in arrays {
                 let rt = udf.return_type(&[array.data_type()]).unwrap();
                 let arg_field = Field::new("arg", array.data_type().clone(), true).into();
-                assert!(matches!(rt, Timestamp(_, Some(_))));
-                let args = datafusion_expr::ScalarFunctionArgs {
+                assert!(matches!(rt, Timestamp(_, None)));
+                let args = ScalarFunctionArgs {
                     args: vec![array.clone()],
                     arg_fields: vec![arg_field],
                     number_rows: 4,
@@ -1042,7 +1667,7 @@ mod tests {
                     _ => panic!("Expected a columnar array"),
                 };
                 let ty = array.data_type();
-                assert!(matches!(ty, Timestamp(_, Some(_))));
+                assert!(matches!(ty, Timestamp(_, None)));
             }
         }
 
@@ -1077,7 +1702,7 @@ mod tests {
                 let rt = udf.return_type(&[array.data_type()]).unwrap();
                 assert!(matches!(rt, Timestamp(_, None)));
                 let arg_field = Field::new("arg", array.data_type().clone(), true).into();
-                let args = datafusion_expr::ScalarFunctionArgs {
+                let args = ScalarFunctionArgs {
                     args: vec![array.clone()],
                     arg_fields: vec![arg_field],
                     number_rows: 5,
diff --git a/datafusion/functions/src/datetime/to_unixtime.rs b/datafusion/functions/src/datetime/to_unixtime.rs
index 42651cd537162..64f201b85ec30 100644
--- a/datafusion/functions/src/datetime/to_unixtime.rs
+++ b/datafusion/functions/src/datetime/to_unixtime.rs
@@ -114,9 +114,11 @@ impl ScalarUDFImpl for ToUnixtimeFunc {
             DataType::Timestamp(_, tz) => arg_args[0]
                 .cast_to(&DataType::Timestamp(TimeUnit::Second, tz), None)?
                 .cast_to(&DataType::Int64, None),
-            DataType::Utf8 => ToTimestampSecondsFunc::new()
-                .invoke_with_args(args)?
-                .cast_to(&DataType::Int64, None),
+            DataType::Utf8 => {
+                ToTimestampSecondsFunc::new_with_config(args.config_options.as_ref())
+                    .invoke_with_args(args)?
+                    .cast_to(&DataType::Int64, None)
+            }
             other => {
                 exec_err!("Unsupported data type {} for function to_unixtime", other)
             }
diff --git a/datafusion/functions/src/macros.rs b/datafusion/functions/src/macros.rs
index f0f3f9efa3786..650d7c31a3307 100644
--- a/datafusion/functions/src/macros.rs
+++ b/datafusion/functions/src/macros.rs
@@ -41,6 +41,17 @@
 /// - `Vec<Expr>` argument (single argument followed by a comma)
 /// - Variable number of `Expr` arguments (zero or more arguments, must be without commas)
 /// - Functions that require config (marked with `@config` prefix)
+///
+/// Note on configuration construction paths:
+/// - The convenience wrappers generated for `@config` functions call the inner
+///   constructor with `ConfigOptions::default()`. These wrappers are intended
+///   primarily for programmatic `Expr` construction and convenience usage.
+/// - When functions are registered in a session, DataFusion will call
+///   `with_updated_config()` to create a `ScalarUDF` instance using the session's
+///   actual `ConfigOptions`. This also happens when configuration changes at runtime
+///   (e.g., via `SET` statements). In short: the macro uses the default config for
+///   convenience constructors; the session config is applied when functions are
+///   registered or when configuration is updated.
 #[macro_export]
 macro_rules! export_functions {
     ($(($FUNC:ident, $DOC:expr, $($arg:tt)*)),*) => {
@@ -59,6 +70,24 @@ macro_rules! export_functions {
         }
     };
 
+    // function that requires config and takes a vector argument
+    (single $FUNC:ident, $DOC:expr, @config $arg:ident,) => {
+        #[doc = $DOC]
+        pub fn $FUNC($arg: Vec<datafusion_expr::Expr>) -> datafusion_expr::Expr {
+            use datafusion_common::config::ConfigOptions;
+            super::$FUNC(&ConfigOptions::default()).call($arg)
+        }
+    };
+
+    // function that requires config and variadic arguments
+    (single $FUNC:ident, $DOC:expr, @config $($arg:ident)*) => {
+        #[doc = $DOC]
+        pub fn $FUNC($($arg: datafusion_expr::Expr),*) -> datafusion_expr::Expr {
+            use datafusion_common::config::ConfigOptions;
+            super::$FUNC(&ConfigOptions::default()).call(vec![$($arg),*])
+        }
+    };
+
     // single vector argument (a single argument followed by a comma)
     (single $FUNC:ident, $DOC:expr, $arg:ident,) => {
         #[doc = $DOC]
diff --git a/datafusion/sqllogictest/test_files/timestamps.slt b/datafusion/sqllogictest/test_files/timestamps.slt
index 3a1a26257b171..8cd27c7b4a42c 100644
--- a/datafusion/sqllogictest/test_files/timestamps.slt
+++ b/datafusion/sqllogictest/test_files/timestamps.slt
@@ -193,6 +193,8 @@ SELECT TIMESTAMPTZ '2000-01-01T01:01:01'
 ----
 2000-01-01T01:01:01Z
 
+statement ok
+RESET datafusion.execution.time_zone
 
 ##########
 ## cast tests
@@ -2256,7 +2258,7 @@ CREATE TABLE foo (time TIMESTAMPTZ) AS VALUES
     ('2020-01-01T03:00:00+05:00')
 
 statement ok
-SET TIME ZONE = '+00'
+RESET datafusion.execution.time_zone
 
 # verify column type
 query T
@@ -2370,7 +2372,7 @@ true true true true true true
 # known issues. currently overflows (expects default precision to be microsecond instead of nanoseconds. Work pending)
 #verify extreme values
 #query PPPPPPPP
-#SELECT to_timestamp(-62125747200), to_timestamp(1926632005177), -62125747200::timestamp, 1926632005177::timestamp, cast(-62125747200 as timestamp), cast(1926632005177 as timestamp)
+#SELECT to_timestamp(-62125747200), to_timestamp(1926632005177), -62125747200::timestamp as t1, 1926632005177::timestamp, cast(-62125747200 as timestamp), cast(1926632005177 as timestamp) as t2
 #----
 #0001-04-25T00:00:00 +63022-07-16T12:59:37 0001-04-25T00:00:00 +63022-07-16T12:59:37 0001-04-25T00:00:00 +63022-07-16T12:59:37
 
@@ -2696,8 +2698,8 @@ SELECT t1.ts, t1.ts + INTERVAL '1' SECOND FROM t1;
 query PT
 SELECT t1.ts::timestamptz, arrow_typeof(t1.ts::timestamptz) FROM t1;
 ----
-2018-07-01T06:00:00Z Timestamp(ns, "+00")
-2018-07-01T07:00:00Z Timestamp(ns, "+00")
+2018-07-01T06:00:00 Timestamp(ns)
+2018-07-01T07:00:00 Timestamp(ns)
 
 query D
 SELECT 0::TIME
@@ -3691,7 +3693,7 @@ SELECT
     arrow_cast(a, 'LargeUtf8')
 FROM (SELECT CAST('2005-09-10 13:31:00 +02:00' AS timestamp with time zone) AS a)
 ----
-Timestamp(ns, "+00") 2005-09-10T11:31:00Z 2005-09-10T11:31:00Z 2005-09-10T11:31:00Z 2005-09-10T11:31:00Z
+Timestamp(ns) 2005-09-10T11:31:00 2005-09-10T11:31:00 2005-09-10T11:31:00 2005-09-10T11:31:00
 
 query P
 SELECT
diff --git a/datafusion/sqllogictest/test_files/to_timestamp_timezone.slt b/datafusion/sqllogictest/test_files/to_timestamp_timezone.slt
new file mode 100644
index 0000000000000..192f103f4e40b
--- /dev/null
+++ b/datafusion/sqllogictest/test_files/to_timestamp_timezone.slt
@@ -0,0 +1,204 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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.
+
+##########
+## to_timestamp timezone tests
+##########
+
+## Reset timezone for other tests
+statement ok
+RESET datafusion.execution.time_zone
+
+## Test 1: Default timezone (None) with naive timestamp
+## Naive timestamps (without explicit timezone) should be interpreted as UTC by default
+query P
+SELECT to_timestamp('2020-09-08T13:42:29');
+----
+2020-09-08T13:42:29
+
+## Test 2: Explicit UTC timezone ('Z' suffix)
+## Explicit timezone should be respected regardless of session timezone
+query P
+SELECT to_timestamp('2020-09-08T13:42:29Z');
+----
+2020-09-08T13:42:29
+
+## Test 3: Explicit timezone offset (+05:00)
+## Explicit offset should be respected - this is 13:42:29+05:00 which is 08:42:29 UTC
+query P
+SELECT to_timestamp('2020-09-08T13:42:29+05:00');
+----
+2020-09-08T08:42:29
+
+## Test 4: Explicit timezone offset without colon (+0500)
+## Should handle offset formats without colons
+query P
+SELECT to_timestamp('2020-09-08T13:42:29+0500');
+----
+2020-09-08T08:42:29
+
+## Test 5: Negative timezone offset
+query P
+SELECT to_timestamp('2020-09-08T13:42:29-03:30');
+----
+2020-09-08T17:12:29
+
+## Test 6: Configure session timezone to America/New_York
+statement ok
+SET datafusion.execution.time_zone = 'America/New_York';
+
+## Test 7: Naive timestamp with configured timezone
+## '2020-09-08T13:42:29' in America/New_York is EDT (UTC-4)
+## So this should become '2020-09-08T13:42:29Z-04:00'
+query P
+SELECT to_timestamp('2020-09-08T13:42:29');
+----
+2020-09-08T13:42:29-04:00
+
+## Test 8: Explicit UTC should be transformed to configured timezone
+query P
+SELECT to_timestamp('2020-09-08T13:42:29Z');
+----
+2020-09-08T09:42:29-04:00
+
+## Test 9: Explicit offset should be transformed to configured timezone
+query P
+SELECT to_timestamp('2020-09-08T13:42:29+05:00');
+----
+2020-09-08T04:42:29-04:00
+
+## Test 10: Check arrow_typeof returns timstamp in configured timezone
+## Result should be Timestamp(Nanosecond, "America/New_York") regardless of input timezone
+query T
+SELECT arrow_typeof(to_timestamp('2020-09-08T13:42:29Z'));
+----
+Timestamp(ns, "America/New_York")
+
+## Test 11: Configure to offset-based timezone
+statement ok
+SET datafusion.execution.time_zone = '+05:30';
+
+## Test 12: Naive timestamp with offset-based configured timezone
+query P
+SELECT to_timestamp('2020-09-08T13:42:29');
+----
+2020-09-08T13:42:29+05:30
+
+## Test 13: Reset to None
+statement ok
+RESET datafusion.execution.time_zone
+
+## Test 14: Naive timestamp
+query P
+SELECT to_timestamp('2020-09-08T13:42:29');
+----
+2020-09-08T13:42:29
+
+query P
+SELECT to_timestamp('2020-09-08T13:42:29Z');
+----
+2020-09-08T13:42:29
+
+query P
+SELECT to_timestamp('2020-09-08T13:42:29+05:00');
+----
+2020-09-08T08:42:29
+
+statement ok
+SET datafusion.execution.time_zone = 'America/New_York';
+
+## Test 15: to_timestamp with format string - naive timestamp with session timezone
+
+query P
+SELECT to_timestamp('2020-09-08 13:42:29', '%Y-%m-%d %H:%M:%S');
+----
+2020-09-08T13:42:29-04:00
+
+## Test 16: to_timestamp with format string - explicit timezone should be respected
+statement ok
+SET datafusion.execution.time_zone = 'UTC';
+
+query P
+SELECT to_timestamp('2020-09-08 13:42:29 +0000', '%Y-%m-%d %H:%M:%S %z');
+----
+2020-09-08T13:42:29Z
+
+query P
+SELECT to_timestamp('2020-09-08 13:42:29 America/Toronto', '%Y-%m-%d %H:%M:%S %Z');
+----
+2020-09-08T17:42:29Z
+
+query error Error parsing timestamp from '2020-09-08 13:42:29America/Toronto' using format '%Y-%m-%d %H:%M:%S%Z': '%Z' is only supported at the end of the format string preceded by a space
+SELECT to_timestamp('2020-09-08 13:42:29America/Toronto', '%Y-%m-%d %H:%M:%S%Z');
+
+## Test 17: Test all precision variants respect timezone
+statement ok
+SET datafusion.execution.time_zone = 'America/New_York';
+
+## to_timestamp_seconds
+query P
+SELECT to_timestamp_seconds('2020-09-08T13:42:29');
+----
+2020-09-08T13:42:29-04:00
+
+## to_timestamp_millis
+query P
+SELECT to_timestamp_millis('2020-09-08T13:42:29.123');
+----
+2020-09-08T13:42:29.123-04:00
+
+## to_timestamp_micros
+query P
+SELECT to_timestamp_micros('2020-09-08T13:42:29.123456');
+----
+2020-09-08T13:42:29.123456-04:00
+
+## to_timestamp_nanos
+query P
+SELECT to_timestamp_nanos('2020-09-08T13:42:29.123456789');
+----
+2020-09-08T13:42:29.123456789-04:00
+
+## test integers
+query T
+select arrow_typeof(to_timestamp_seconds(61))
+----
+Timestamp(s, "America/New_York")
+
+query T
+select arrow_typeof(to_timestamp_millis(61))
+----
+Timestamp(ms, "America/New_York")
+
+query T
+select arrow_typeof(to_timestamp_micros(61))
+----
+Timestamp(µs, "America/New_York")
+
+query T
+select arrow_typeof(to_timestamp_nanos(61))
+----
+Timestamp(ns, "America/New_York")
+
+query T
+select arrow_typeof(to_timestamp(61))
+----
+Timestamp(ns, "America/New_York")
+
+## Reset timezone for other tests
+statement ok
+RESET datafusion.execution.time_zone
\ No newline at end of file
diff --git a/docs/source/user-guide/sql/scalar_functions.md b/docs/source/user-guide/sql/scalar_functions.md
index c5380b2e84b77..720f06ff768ab 100644
--- a/docs/source/user-guide/sql/scalar_functions.md
+++ b/docs/source/user-guide/sql/scalar_functions.md
@@ -2775,9 +2775,19 @@ FROM (
 
 ### `to_timestamp`
 
-Converts a value to a timestamp (`YYYY-MM-DDT00:00:00Z`). Supports strings, integer, unsigned integer, and double types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00') if no [Chrono formats] are provided. Integers, unsigned integers, and doubles are interpreted as seconds since the unix epoch (`1970-01-01T00:00:00Z`). Returns the corresponding timestamp.
+Converts a value to a timestamp (`YYYY-MM-DDT00:00:00.000000<TZ>`) in the session time zone. Supports strings,
+integer, unsigned integer, and double types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00')
+if no [Chrono formats] are provided. Strings that parse without a time zone are treated as if they are in the
+session time zone, or UTC if no session time zone is set.
+Integers, unsigned integers, and doubles are interpreted as seconds since the unix epoch (`1970-01-01T00:00:00Z`).
 
-Note: `to_timestamp` returns `Timestamp(ns)`. The supported range for integer input is between `-9223372037` and `9223372036`. Supported range for string input is between `1677-09-21T00:12:44.0` and `2262-04-11T23:47:16.0`. Please use `to_timestamp_seconds` for the input outside of supported bounds.
+Note: `to_timestamp` returns `Timestamp(ns, TimeZone)` where the time zone is the session time zone. The supported range
+for integer input is between`-9223372037` and `9223372036`. Supported range for string input is between
+`1677-09-21T00:12:44.0` and `2262-04-11T23:47:16.0`. Please use `to_timestamp_seconds`
+for the input outside of supported bounds.
+
+The session time zone can be set using the statement `SET TIMEZONE = 'desired time zone'`.
+The time zone can be a value like +00:00, 'Europe/London' etc.
 
 ```sql
 to_timestamp(expression[, ..., format_n])
@@ -2786,7 +2796,11 @@ to_timestamp(expression[, ..., format_n])
 #### Arguments
 
 - **expression**: Expression to operate on. Can be a constant, column, or function, and any combination of arithmetic operators.
-- **format_n**: Optional [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) strings to use to parse the expression. Formats will be tried in the order they appear with the first successful one being returned. If none of the formats successfully parse the expression an error will be returned.
+- **format_n**:
+  Optional [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) strings to use to parse the expression.
+  Formats will be tried in the order they appear with the first successful one being returned. If none of the formats successfully
+  parse the expression an error will be returned. Note: parsing of named timezones (e.g. 'America/New_York') using %Z is
+  only supported at the end of the string preceded by a space.
 
 #### Example
 
@@ -2809,7 +2823,14 @@ Additional examples can be found [here](https://github.com/apache/datafusion/blo
 
 ### `to_timestamp_micros`
 
-Converts a value to a timestamp (`YYYY-MM-DDT00:00:00.000000Z`). Supports strings, integer, and unsigned integer types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00') if no [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html)s are provided. Integers and unsigned integers are interpreted as microseconds since the unix epoch (`1970-01-01T00:00:00Z`) Returns the corresponding timestamp.
+Converts a value to a timestamp (`YYYY-MM-DDT00:00:00.000000<TZ>`) in the session time zone. Supports strings,
+integer, unsigned integer, and double types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00')
+if no [Chrono formats] are provided. Strings that parse without a time zone are treated as if they are in the
+session time zone, or UTC if no session time zone is set.
+Integers, unsigned integers, and doubles are interpreted as microseconds since the unix epoch (`1970-01-01T00:00:00Z`).
+
+The session time zone can be set using the statement `SET TIMEZONE = 'desired time zone'`.
+The time zone can be a value like +00:00, 'Europe/London' etc.
 
 ```sql
 to_timestamp_micros(expression[, ..., format_n])
@@ -2818,7 +2839,11 @@ to_timestamp_micros(expression[, ..., format_n])
 #### Arguments
 
 - **expression**: Expression to operate on. Can be a constant, column, or function, and any combination of arithmetic operators.
-- **format_n**: Optional [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) strings to use to parse the expression. Formats will be tried in the order they appear with the first successful one being returned. If none of the formats successfully parse the expression an error will be returned.
+- **format_n**:
+  Optional [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) strings to use to parse the expression.
+  Formats will be tried in the order they appear with the first successful one being returned. If none of the formats successfully
+  parse the expression an error will be returned. Note: parsing of named timezones (e.g. 'America/New_York') using %Z is
+  only supported at the end of the string preceded by a space.
 
 #### Example
 
@@ -2841,7 +2866,14 @@ Additional examples can be found [here](https://github.com/apache/datafusion/blo
 
 ### `to_timestamp_millis`
 
-Converts a value to a timestamp (`YYYY-MM-DDT00:00:00.000Z`). Supports strings, integer, and unsigned integer types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00') if no [Chrono formats](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) are provided. Integers and unsigned integers are interpreted as milliseconds since the unix epoch (`1970-01-01T00:00:00Z`). Returns the corresponding timestamp.
+Converts a value to a timestamp (`YYYY-MM-DDT00:00:00.000<TZ>`) in the session time zone. Supports strings,
+integer, unsigned integer, and double types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00')
+if no [Chrono formats] are provided. Strings that parse without a time zone are treated as if they are in the
+session time zone, or UTC if no session time zone is set.
+Integers, unsigned integers, and doubles are interpreted as milliseconds since the unix epoch (`1970-01-01T00:00:00Z`).
+
+The session time zone can be set using the statement `SET TIMEZONE = 'desired time zone'`.
+The time zone can be a value like +00:00, 'Europe/London' etc.
 
 ```sql
 to_timestamp_millis(expression[, ..., format_n])
@@ -2850,7 +2882,11 @@ to_timestamp_millis(expression[, ..., format_n])
 #### Arguments
 
 - **expression**: Expression to operate on. Can be a constant, column, or function, and any combination of arithmetic operators.
-- **format_n**: Optional [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) strings to use to parse the expression. Formats will be tried in the order they appear with the first successful one being returned. If none of the formats successfully parse the expression an error will be returned.
+- **format_n**:
+  Optional [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) strings to use to parse the expression.
+  Formats will be tried in the order they appear with the first successful one being returned. If none of the formats successfully
+  parse the expression an error will be returned. Note: parsing of named timezones (e.g. 'America/New_York') using %Z is
+  only supported at the end of the string preceded by a space.
 
 #### Example
 
@@ -2873,7 +2909,13 @@ Additional examples can be found [here](https://github.com/apache/datafusion/blo
 
 ### `to_timestamp_nanos`
 
-Converts a value to a timestamp (`YYYY-MM-DDT00:00:00.000000000Z`). Supports strings, integer, and unsigned integer types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00') if no [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html)s are provided. Integers and unsigned integers are interpreted as nanoseconds since the unix epoch (`1970-01-01T00:00:00Z`). Returns the corresponding timestamp.
+Converts a value to a timestamp (`YYYY-MM-DDT00:00:00.000000000Z<TZ>`) in the session time zone. Supports strings,
+integer, unsigned integer, and double types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00')
+if no [Chrono formats] are provided. Strings that parse without a time zone are treated as if they are in the
+session time zone. Integers, unsigned integers, and doubles are interpreted as nanoseconds since the unix epoch (`1970-01-01T00:00:00Z`).
+
+The session time zone can be set using the statement `SET TIMEZONE = 'desired time zone'`.
+The time zone can be a value like +00:00, 'Europe/London' etc.
 
 ```sql
 to_timestamp_nanos(expression[, ..., format_n])
@@ -2882,7 +2924,11 @@ to_timestamp_nanos(expression[, ..., format_n])
 #### Arguments
 
 - **expression**: Expression to operate on. Can be a constant, column, or function, and any combination of arithmetic operators.
-- **format_n**: Optional [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) strings to use to parse the expression. Formats will be tried in the order they appear with the first successful one being returned. If none of the formats successfully parse the expression an error will be returned.
+- **format_n**:
+  Optional [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) strings to use to parse the expression.
+  Formats will be tried in the order they appear with the first successful one being returned. If none of the formats successfully
+  parse the expression an error will be returned. Note: parsing of named timezones (e.g. 'America/New_York') using %Z is
+  only supported at the end of the string preceded by a space.
 
 #### Example
 
@@ -2905,7 +2951,14 @@ Additional examples can be found [here](https://github.com/apache/datafusion/blo
 
 ### `to_timestamp_seconds`
 
-Converts a value to a timestamp (`YYYY-MM-DDT00:00:00.000Z`). Supports strings, integer, and unsigned integer types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00') if no [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html)s are provided. Integers and unsigned integers are interpreted as seconds since the unix epoch (`1970-01-01T00:00:00Z`). Returns the corresponding timestamp.
+Converts a value to a timestamp (`YYYY-MM-DDT00:00:00<TZ>`) in the session time zone. Supports strings,
+integer, unsigned integer, and double types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00')
+if no [Chrono formats] are provided. Strings that parse without a time zone are treated as if they are in the
+session time zone, or UTC if no session time zone is set.
+Integers, unsigned integers, and doubles are interpreted as seconds since the unix epoch (`1970-01-01T00:00:00Z`).
+
+The session time zone can be set using the statement `SET TIMEZONE = 'desired time zone'`.
+The time zone can be a value like +00:00, 'Europe/London' etc.
 
 ```sql
 to_timestamp_seconds(expression[, ..., format_n])
@@ -2914,7 +2967,11 @@ to_timestamp_seconds(expression[, ..., format_n])
 #### Arguments
 
 - **expression**: Expression to operate on. Can be a constant, column, or function, and any combination of arithmetic operators.
-- **format_n**: Optional [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) strings to use to parse the expression. Formats will be tried in the order they appear with the first successful one being returned. If none of the formats successfully parse the expression an error will be returned.
+- **format_n**:
+  Optional [Chrono format](https://docs.rs/chrono/latest/chrono/format/strftime/index.html) strings to use to parse the expression.
+  Formats will be tried in the order they appear with the first successful one being returned. If none of the formats successfully
+  parse the expression an error will be returned. Note: parsing of named timezones (e.g. 'America/New_York') using %Z is
+  only supported at the end of the string preceded by a space.
 
 #### Example
 

From 5c4eaf7b68c30258ec5d3d3b67b61af19120d843 Mon Sep 17 00:00:00 2001
From: Bruce Ritchie <bruce.ritchie@gmail.com>
Date: Thu, 4 Dec 2025 10:23:06 -0500
Subject: [PATCH 02/11] Updates from code review.

---
 datafusion/functions/Cargo.toml               |  4 +--
 datafusion/functions/src/datetime/common.rs   | 27 ++++++++++---------
 .../functions/src/datetime/to_timestamp.rs    |  2 +-
 .../test_files/to_timestamp_timezone.slt      |  2 +-
 .../source/user-guide/sql/scalar_functions.md |  2 +-
 5 files changed, 20 insertions(+), 17 deletions(-)

diff --git a/datafusion/functions/Cargo.toml b/datafusion/functions/Cargo.toml
index 01c78ebc5f8bd..48b29fa708404 100644
--- a/datafusion/functions/Cargo.toml
+++ b/datafusion/functions/Cargo.toml
@@ -40,7 +40,7 @@ workspace = true
 [features]
 crypto_expressions = ["md-5", "sha2", "blake2", "blake3"]
 # enable datetime functions
-datetime_expressions = []
+datetime_expressions = ["chrono-tz"]
 # Enable encoding by default so the doctests work. In general don't automatically enable all packages.
 default = [
     "datetime_expressions",
@@ -71,7 +71,7 @@ base64 = { version = "0.22", optional = true }
 blake2 = { version = "^0.10.2", optional = true }
 blake3 = { version = "1.8", optional = true }
 chrono = { workspace = true }
-chrono-tz = "0.10.4"
+chrono-tz = { version = "0.10.4", optional = true }
 datafusion-common = { workspace = true }
 datafusion-doc = { workspace = true }
 datafusion-execution = { workspace = true }
diff --git a/datafusion/functions/src/datetime/common.rs b/datafusion/functions/src/datetime/common.rs
index 338bf73254b94..83121c5930585 100644
--- a/datafusion/functions/src/datetime/common.rs
+++ b/datafusion/functions/src/datetime/common.rs
@@ -116,20 +116,23 @@ pub(crate) fn string_to_datetime_formatted<T: TimeZone>(
     // custom parser (or switching to Jiff)
     let tz: Option<chrono_tz::Tz> = if format.ends_with(" %Z") {
         // grab the string after the last space as the named timezone
-        let parts: Vec<&str> = datetime_str.rsplitn(2, ' ').collect();
-        let timezone_name = parts[0];
-        datetime_str = parts[1];
-
-        // attempt to parse the timezone name
-        let result: Result<chrono_tz::Tz, chrono_tz::ParseError> = timezone_name.parse();
-        let Ok(tz) = result else {
-            return Err(err(&result.unwrap_err().to_string()));
-        };
+        if let Some((dt_str, timezone_name)) = datetime_str.rsplit_once(' ') {
+            datetime_str = dt_str;
+
+            // attempt to parse the timezone name
+            let result: Result<chrono_tz::Tz, chrono_tz::ParseError> =
+                timezone_name.parse();
+            let Ok(tz) = result else {
+                return Err(err(&result.unwrap_err().to_string()));
+            };
 
-        // successfully parsed the timezone name, remove the ' %Z' from the format
-        format = format.trim_end_matches(" %Z");
+            // successfully parsed the timezone name, remove the ' %Z' from the format
+            format = &format[..format.len() - 3];
 
-        Some(tz)
+            Some(tz)
+        } else {
+            None
+        }
     } else if format.contains("%Z") {
         return Err(err(
             "'%Z' is only supported at the end of the format string preceded by a space",
diff --git a/datafusion/functions/src/datetime/to_timestamp.rs b/datafusion/functions/src/datetime/to_timestamp.rs
index ce5ed3a4a92a4..cd8659ee6efe6 100644
--- a/datafusion/functions/src/datetime/to_timestamp.rs
+++ b/datafusion/functions/src/datetime/to_timestamp.rs
@@ -239,7 +239,7 @@ pub struct ToTimestampMicrosFunc {
 #[user_doc(
     doc_section(label = "Time and Date Functions"),
     description = r#"
-Converts a value to a timestamp (`YYYY-MM-DDT00:00:00.000000000Z<TZ>`) in the session time zone. Supports strings,
+Converts a value to a timestamp (`YYYY-MM-DDT00:00:00.000000000<TZ>`) in the session time zone. Supports strings,
 integer, unsigned integer, and double types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00')
 if no [Chrono formats] are provided. Strings that parse without a time zone are treated as if they are in the
 session time zone. Integers, unsigned integers, and doubles are interpreted as nanoseconds since the unix epoch (`1970-01-01T00:00:00Z`).
diff --git a/datafusion/sqllogictest/test_files/to_timestamp_timezone.slt b/datafusion/sqllogictest/test_files/to_timestamp_timezone.slt
index 192f103f4e40b..ee7fab119749d 100644
--- a/datafusion/sqllogictest/test_files/to_timestamp_timezone.slt
+++ b/datafusion/sqllogictest/test_files/to_timestamp_timezone.slt
@@ -63,7 +63,7 @@ SET datafusion.execution.time_zone = 'America/New_York';
 
 ## Test 7: Naive timestamp with configured timezone
 ## '2020-09-08T13:42:29' in America/New_York is EDT (UTC-4)
-## So this should become '2020-09-08T13:42:29Z-04:00'
+## So this should become '2020-09-08T13:42:29-04:00'
 query P
 SELECT to_timestamp('2020-09-08T13:42:29');
 ----
diff --git a/docs/source/user-guide/sql/scalar_functions.md b/docs/source/user-guide/sql/scalar_functions.md
index 720f06ff768ab..27390b94e2877 100644
--- a/docs/source/user-guide/sql/scalar_functions.md
+++ b/docs/source/user-guide/sql/scalar_functions.md
@@ -2909,7 +2909,7 @@ Additional examples can be found [here](https://github.com/apache/datafusion/blo
 
 ### `to_timestamp_nanos`
 
-Converts a value to a timestamp (`YYYY-MM-DDT00:00:00.000000000Z<TZ>`) in the session time zone. Supports strings,
+Converts a value to a timestamp (`YYYY-MM-DDT00:00:00.000000000<TZ>`) in the session time zone. Supports strings,
 integer, unsigned integer, and double types as input. Strings are parsed as RFC3339 (e.g. '2023-07-20T05:44:00')
 if no [Chrono formats] are provided. Strings that parse without a time zone are treated as if they are in the
 session time zone. Integers, unsigned integers, and doubles are interpreted as nanoseconds since the unix epoch (`1970-01-01T00:00:00Z`).

From 15096a6c13e602d580f056828a5e5fecc0d31051 Mon Sep 17 00:00:00 2001
From: Bruce Ritchie <bruce.ritchie@gmail.com>
Date: Fri, 5 Dec 2025 09:19:15 -0500
Subject: [PATCH 03/11] Extract parsing of UTC to a static.

---
 datafusion/functions/src/datetime/common.rs | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/datafusion/functions/src/datetime/common.rs b/datafusion/functions/src/datetime/common.rs
index 83121c5930585..ce24cc67bd09f 100644
--- a/datafusion/functions/src/datetime/common.rs
+++ b/datafusion/functions/src/datetime/common.rs
@@ -16,7 +16,7 @@
 // under the License.
 
 use std::marker::PhantomData;
-use std::sync::Arc;
+use std::sync::{Arc, LazyLock};
 
 use arrow::array::timezone::Tz;
 use arrow::array::{
@@ -42,6 +42,8 @@ use num_traits::{PrimInt, ToPrimitive};
 /// Error message if nanosecond conversion request beyond supported interval
 const ERR_NANOSECONDS_NOT_SUPPORTED: &str = "The dates that can be represented as nanoseconds have to be between 1677-09-21T00:12:44.0 and 2262-04-11T23:47:16.854775804";
 
+static UTC: LazyLock<Tz> = LazyLock::new(|| "UTC".parse().expect("UTC is always valid"));
+
 #[expect(unused)]
 /// Calls string_to_timestamp_nanos and converts the error type
 pub(crate) fn string_to_timestamp_nanos_shim(s: &str) -> Result<i64> {
@@ -52,7 +54,7 @@ pub(crate) fn string_to_timestamp_nanos_with_timezone(
     timezone: &Option<Tz>,
     s: &str,
 ) -> Result<i64> {
-    let tz = timezone.unwrap_or("UTC".parse()?);
+    let tz = timezone.unwrap_or(*UTC);
     let dt = string_to_datetime(&tz, s)?;
     let parsed = dt
         .timestamp_nanos_opt()
@@ -220,8 +222,7 @@ pub(crate) fn string_to_timestamp_nanos_formatted_with_timezone(
     s: &str,
     format: &str,
 ) -> Result<i64, DataFusionError> {
-    let dt =
-        string_to_datetime_formatted(&timezone.unwrap_or("UTC".parse()?), s, format)?;
+    let dt = string_to_datetime_formatted(&timezone.unwrap_or(*UTC), s, format)?;
     let parsed = dt
         .timestamp_nanos_opt()
         .ok_or_else(|| exec_datafusion_err!("{ERR_NANOSECONDS_NOT_SUPPORTED}"))?;

From 7dded4276ef4d8e41cf37f4c20c10da89ee28364 Mon Sep 17 00:00:00 2001
From: Bruce Ritchie <bruce.ritchie@gmail.com>
Date: Fri, 12 Dec 2025 07:33:18 -0500
Subject: [PATCH 04/11] Updates from merge.

---
 datafusion/functions/src/datetime/common.rs       | 2 +-
 datafusion/functions/src/datetime/to_timestamp.rs | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/datafusion/functions/src/datetime/common.rs b/datafusion/functions/src/datetime/common.rs
index 7f41f165cb1b7..b4cddc9945673 100644
--- a/datafusion/functions/src/datetime/common.rs
+++ b/datafusion/functions/src/datetime/common.rs
@@ -33,7 +33,7 @@ use chrono::format::{Parsed, StrftimeItems, parse};
 use chrono::{DateTime, TimeZone, Utc};
 use datafusion_common::cast::as_generic_string_array;
 use datafusion_common::{
-    DataFusionError, Result, ScalarType, ScalarValue, exec_datafusion_err, exec_err,
+    DataFusionError, Result, ScalarValue, exec_datafusion_err, exec_err,
     internal_datafusion_err, unwrap_or_internal_err,
 };
 use datafusion_expr::ColumnarValue;
diff --git a/datafusion/functions/src/datetime/to_timestamp.rs b/datafusion/functions/src/datetime/to_timestamp.rs
index 181bed594d53a..728413142304b 100644
--- a/datafusion/functions/src/datetime/to_timestamp.rs
+++ b/datafusion/functions/src/datetime/to_timestamp.rs
@@ -19,8 +19,8 @@ use std::any::Any;
 use std::sync::Arc;
 
 use crate::datetime::common::*;
-use arrow::array::timezone::Tz;
 use arrow::array::Float64Array;
+use arrow::array::timezone::Tz;
 use arrow::datatypes::DataType::*;
 use arrow::datatypes::TimeUnit::{Microsecond, Millisecond, Nanosecond, Second};
 use arrow::datatypes::{

From 1ea061c78e2ec3c4ac32814ac192e913d3009e19 Mon Sep 17 00:00:00 2001
From: Bruce Ritchie <bruce.ritchie@gmail.com>
Date: Fri, 12 Dec 2025 08:17:24 -0500
Subject: [PATCH 05/11] remove unused functions.

---
 datafusion/functions/src/datetime/common.rs | 26 +++------------------
 1 file changed, 3 insertions(+), 23 deletions(-)

diff --git a/datafusion/functions/src/datetime/common.rs b/datafusion/functions/src/datetime/common.rs
index b4cddc9945673..b6555c4694967 100644
--- a/datafusion/functions/src/datetime/common.rs
+++ b/datafusion/functions/src/datetime/common.rs
@@ -23,9 +23,7 @@ use arrow::array::{
     Array, ArrowPrimitiveType, AsArray, GenericStringArray, PrimitiveArray,
     StringArrayType, StringViewArray,
 };
-use arrow::compute::kernels::cast_utils::{
-    string_to_datetime, string_to_timestamp_nanos,
-};
+use arrow::compute::kernels::cast_utils::string_to_datetime;
 use arrow::datatypes::{DataType, TimeUnit};
 use arrow_buffer::ArrowNativeType;
 use chrono::LocalResult::Single;
@@ -44,12 +42,6 @@ const ERR_NANOSECONDS_NOT_SUPPORTED: &str = "The dates that can be represented a
 
 static UTC: LazyLock<Tz> = LazyLock::new(|| "UTC".parse().expect("UTC is always valid"));
 
-#[expect(unused)]
-/// Calls string_to_timestamp_nanos and converts the error type
-pub(crate) fn string_to_timestamp_nanos_shim(s: &str) -> Result<i64> {
-    string_to_timestamp_nanos(s).map_err(|e| e.into())
-}
-
 pub(crate) fn string_to_timestamp_nanos_with_timezone(
     timezone: &Option<Tz>,
     s: &str,
@@ -179,7 +171,7 @@ pub(crate) fn string_to_datetime_formatted<T: TimeZone>(
 }
 
 /// Accepts a string with a `chrono` format and converts it to a
-/// nanosecond precision timestamp.
+/// nanosecond precision timestamp relative to the provided `timezone`.
 ///
 /// See [`chrono::format::strftime`] for the full set of supported formats.
 ///
@@ -205,18 +197,6 @@ pub(crate) fn string_to_datetime_formatted<T: TimeZone>(
 ///
 /// [`chrono::format::strftime`]: https://docs.rs/chrono/latest/chrono/format/strftime/index.html
 #[inline]
-#[expect(unused)]
-pub(crate) fn string_to_timestamp_nanos_formatted(
-    s: &str,
-    format: &str,
-) -> Result<i64, DataFusionError> {
-    string_to_datetime_formatted(&Utc, s, format)?
-        .naive_utc()
-        .and_utc()
-        .timestamp_nanos_opt()
-        .ok_or_else(|| exec_datafusion_err!("{ERR_NANOSECONDS_NOT_SUPPORTED}"))
-}
-
 pub(crate) fn string_to_timestamp_nanos_formatted_with_timezone(
     timezone: &Option<Tz>,
     s: &str,
@@ -231,7 +211,7 @@ pub(crate) fn string_to_timestamp_nanos_formatted_with_timezone(
 }
 
 /// Accepts a string with a `chrono` format and converts it to a
-/// millisecond precision timestamp.
+/// millisecond precision timestamp relative to the provided `timezone`.
 ///
 /// See [`chrono::format::strftime`] for the full set of supported formats.
 ///

From e89d0bdd7a85df20cc7b92d1ed259e99767d8b21 Mon Sep 17 00:00:00 2001
From: Bruce Ritchie <bruce.ritchie@gmail.com>
Date: Wed, 17 Dec 2025 09:54:44 -0500
Subject: [PATCH 06/11] PR review feedback.

---
 datafusion/functions/benches/to_timestamp.rs                | 5 ++++-
 datafusion/functions/src/datetime/common.rs                 | 6 +++---
 .../sqllogictest/test_files/to_timestamp_timezone.slt       | 2 +-
 3 files changed, 8 insertions(+), 5 deletions(-)

diff --git a/datafusion/functions/benches/to_timestamp.rs b/datafusion/functions/benches/to_timestamp.rs
index 3a3420f9481f4..ed865fa6e8d50 100644
--- a/datafusion/functions/benches/to_timestamp.rs
+++ b/datafusion/functions/benches/to_timestamp.rs
@@ -114,7 +114,10 @@ fn criterion_benchmark(c: &mut Criterion) {
         Field::new("f", DataType::Timestamp(TimeUnit::Nanosecond, None), true).into();
     let arg_field = Field::new("a", DataType::Utf8, false).into();
     let arg_fields = vec![arg_field];
-    let config_options = Arc::new(ConfigOptions::default());
+    let mut options = ConfigOptions::default();
+    options.execution.time_zone = Some("UTC".into());
+    let config_options = Arc::new(options);
+
     let to_timestamp_udf = to_timestamp(config_options.as_ref());
 
     c.bench_function("to_timestamp_no_formats_utf8", |b| {
diff --git a/datafusion/functions/src/datetime/common.rs b/datafusion/functions/src/datetime/common.rs
index b6555c4694967..3afbe0606eb93 100644
--- a/datafusion/functions/src/datetime/common.rs
+++ b/datafusion/functions/src/datetime/common.rs
@@ -46,8 +46,8 @@ pub(crate) fn string_to_timestamp_nanos_with_timezone(
     timezone: &Option<Tz>,
     s: &str,
 ) -> Result<i64> {
-    let tz = timezone.unwrap_or(*UTC);
-    let dt = string_to_datetime(&tz, s)?;
+    let tz = timezone.as_ref().unwrap_or(&UTC);
+    let dt = string_to_datetime(tz, s)?;
     let parsed = dt
         .timestamp_nanos_opt()
         .ok_or_else(|| exec_datafusion_err!("{ERR_NANOSECONDS_NOT_SUPPORTED}"))?;
@@ -202,7 +202,7 @@ pub(crate) fn string_to_timestamp_nanos_formatted_with_timezone(
     s: &str,
     format: &str,
 ) -> Result<i64, DataFusionError> {
-    let dt = string_to_datetime_formatted(&timezone.unwrap_or(*UTC), s, format)?;
+    let dt = string_to_datetime_formatted(timezone.as_ref().unwrap_or(&UTC), s, format)?;
     let parsed = dt
         .timestamp_nanos_opt()
         .ok_or_else(|| exec_datafusion_err!("{ERR_NANOSECONDS_NOT_SUPPORTED}"))?;
diff --git a/datafusion/sqllogictest/test_files/to_timestamp_timezone.slt b/datafusion/sqllogictest/test_files/to_timestamp_timezone.slt
index ee7fab119749d..d48e41d1204de 100644
--- a/datafusion/sqllogictest/test_files/to_timestamp_timezone.slt
+++ b/datafusion/sqllogictest/test_files/to_timestamp_timezone.slt
@@ -201,4 +201,4 @@ Timestamp(ns, "America/New_York")
 
 ## Reset timezone for other tests
 statement ok
-RESET datafusion.execution.time_zone
\ No newline at end of file
+RESET datafusion.execution.time_zone

From f4e0934a1e212e13861c20a0954e43b1572a7a08 Mon Sep 17 00:00:00 2001
From: Bruce Ritchie <bruce.ritchie@gmail.com>
Date: Wed, 17 Dec 2025 10:15:46 -0500
Subject: [PATCH 07/11] PR review feedback.

---
 datafusion/functions/src/datetime/common.rs | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/datafusion/functions/src/datetime/common.rs b/datafusion/functions/src/datetime/common.rs
index 3afbe0606eb93..293d543da143b 100644
--- a/datafusion/functions/src/datetime/common.rs
+++ b/datafusion/functions/src/datetime/common.rs
@@ -108,9 +108,9 @@ pub(crate) fn string_to_datetime_formatted<T: TimeZone>(
     // Note that %+ handles 'Z' at the end of the string without a space. This code doesn't
     // handle named timezones with no preceding space since that would require writing a
     // custom parser (or switching to Jiff)
-    let tz: Option<chrono_tz::Tz> = if format.ends_with(" %Z") {
+    let tz: Option<chrono_tz::Tz> = if format.trim_end().ends_with(" %Z") {
         // grab the string after the last space as the named timezone
-        if let Some((dt_str, timezone_name)) = datetime_str.rsplit_once(' ') {
+        if let Some((dt_str, timezone_name)) = datetime_str.trim_end().rsplit_once(' ') {
             datetime_str = dt_str;
 
             // attempt to parse the timezone name

From 4a9174325213e44116698c18a98d2c1f16c2ef96 Mon Sep 17 00:00:00 2001
From: Bruce Ritchie <bruce.ritchie@gmail.com>
Date: Sat, 20 Dec 2025 11:20:24 -0500
Subject: [PATCH 08/11] Add documentation.

---
 datafusion/functions/src/datetime/common.rs | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/datafusion/functions/src/datetime/common.rs b/datafusion/functions/src/datetime/common.rs
index 293d543da143b..c901418453fe7 100644
--- a/datafusion/functions/src/datetime/common.rs
+++ b/datafusion/functions/src/datetime/common.rs
@@ -42,6 +42,20 @@ const ERR_NANOSECONDS_NOT_SUPPORTED: &str = "The dates that can be represented a
 
 static UTC: LazyLock<Tz> = LazyLock::new(|| "UTC".parse().expect("UTC is always valid"));
 
+/// Converts a string representation of a date‑time into a timestamp expressed in
+/// nanoseconds since the Unix epoch.
+///
+/// This helper is a thin wrapper around the more general `string_to_datetime`
+/// function. It accepts an optional `timezone` which, if `None`, defaults to
+/// Coordinated Universal Time (UTC). The string `s` must contain a valid
+/// date‑time format that can be parsed by the underlying chrono parser.
+///
+/// # Return Value
+///
+/// * `Ok(i64)` – The number of nanoseconds since `1970‑01‑01T00:00:00Z`.
+/// * `Err(DataFusionError)` – If the string cannot be parsed, the parsed
+///   value is out of range (between 1677-09-21T00:12:44.0 and 2262-04-11T23:47:16.854775804)
+///   or the parsed value does not correspond to an unambiguous time.
 pub(crate) fn string_to_timestamp_nanos_with_timezone(
     timezone: &Option<Tz>,
     s: &str,

From 8d9cb2e0882f6ddbe59cd6cf8879bff6db7f21b2 Mon Sep 17 00:00:00 2001
From: Bruce Ritchie <bruce.ritchie@gmail.com>
Date: Sat, 20 Dec 2025 17:21:35 -0500
Subject: [PATCH 09/11] Refactored ScalarDataType out of the PR.

---
 datafusion/functions/src/datetime/common.rs   | 65 ++++++-------------
 datafusion/functions/src/datetime/to_date.rs  |  8 +--
 .../functions/src/datetime/to_timestamp.rs    |  8 +--
 3 files changed, 28 insertions(+), 53 deletions(-)

diff --git a/datafusion/functions/src/datetime/common.rs b/datafusion/functions/src/datetime/common.rs
index c901418453fe7..2db64beafa9b7 100644
--- a/datafusion/functions/src/datetime/common.rs
+++ b/datafusion/functions/src/datetime/common.rs
@@ -15,7 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::marker::PhantomData;
 use std::sync::{Arc, LazyLock};
 
 use arrow::array::timezone::Tz;
@@ -23,6 +22,7 @@ use arrow::array::{
     Array, ArrowPrimitiveType, AsArray, GenericStringArray, PrimitiveArray,
     StringArrayType, StringViewArray,
 };
+use arrow::compute::DecimalCast;
 use arrow::compute::kernels::cast_utils::string_to_datetime;
 use arrow::datatypes::{DataType, TimeUnit};
 use arrow_buffer::ArrowNativeType;
@@ -35,7 +35,6 @@ use datafusion_common::{
     internal_datafusion_err, unwrap_or_internal_err,
 };
 use datafusion_expr::ColumnarValue;
-use num_traits::{PrimInt, ToPrimitive};
 
 /// Error message if nanosecond conversion request beyond supported interval
 const ERR_NANOSECONDS_NOT_SUPPORTED: &str = "The dates that can be represented as nanoseconds have to be between 1677-09-21T00:12:44.0 and 2262-04-11T23:47:16.854775804";
@@ -248,50 +247,14 @@ pub(crate) fn string_to_timestamp_millis_formatted(s: &str, format: &str) -> Res
         .timestamp_millis())
 }
 
-pub(crate) struct ScalarDataType<T: PrimInt> {
-    data_type: DataType,
-    _marker: PhantomData<T>,
-}
-
-impl<T: PrimInt> ScalarDataType<T> {
-    pub(crate) fn new(dt: DataType) -> Self {
-        Self {
-            data_type: dt,
-            _marker: PhantomData,
-        }
-    }
-
-    fn scalar(&self, r: Option<i64>) -> Result<ScalarValue> {
-        match &self.data_type {
-            DataType::Date32 => Ok(ScalarValue::Date32(r.and_then(|v| v.to_i32()))),
-            DataType::Timestamp(u, tz) => match u {
-                TimeUnit::Second => Ok(ScalarValue::TimestampSecond(r, tz.clone())),
-                TimeUnit::Millisecond => {
-                    Ok(ScalarValue::TimestampMillisecond(r, tz.clone()))
-                }
-                TimeUnit::Microsecond => {
-                    Ok(ScalarValue::TimestampMicrosecond(r, tz.clone()))
-                }
-                TimeUnit::Nanosecond => {
-                    Ok(ScalarValue::TimestampNanosecond(r, tz.clone()))
-                }
-            },
-            t => Err(internal_datafusion_err!(
-                "Unsupported data type for ScalarDataType<T>: {t:?}"
-            )),
-        }
-    }
-}
-
-pub(crate) fn handle<O, F, T>(
+pub(crate) fn handle<O, F>(
     args: &[ColumnarValue],
     op: F,
     name: &str,
-    sdt: &ScalarDataType<T>,
+    dt: &DataType,
 ) -> Result<ColumnarValue>
 where
     O: ArrowPrimitiveType,
-    T: PrimInt,
     F: Fn(&str) -> Result<O::Native>,
 {
     match &args[0] {
@@ -323,7 +286,7 @@ where
                     .map(|x| op(x))
                     .transpose()?
                     .and_then(|v| v.to_i64());
-                let s = sdt.scalar(result)?;
+                let s = scalar_value(dt, result)?;
                 Ok(ColumnarValue::Scalar(s))
             }
             _ => exec_err!("Unsupported data type {scalar:?} for function {name}"),
@@ -334,18 +297,17 @@ where
 // Given a function that maps a `&str`, `&str` to an arrow native type,
 // returns a `ColumnarValue` where the function is applied to either a `ArrayRef` or `ScalarValue`
 // depending on the `args`'s variant.
-pub(crate) fn handle_multiple<O, F, M, T>(
+pub(crate) fn handle_multiple<O, F, M>(
     args: &[ColumnarValue],
     op: F,
     op2: M,
     name: &str,
-    sdt: &ScalarDataType<T>,
+    dt: &DataType,
 ) -> Result<ColumnarValue>
 where
     O: ArrowPrimitiveType,
     F: Fn(&str, &str) -> Result<O::Native>,
     M: Fn(O::Native) -> O::Native,
-    T: PrimInt,
 {
     match &args[0] {
         ColumnarValue::Array(a) => match a.data_type() {
@@ -413,7 +375,7 @@ where
                         match op(a, s.as_str()) {
                             Ok(r) => {
                                 let result = op2(r).to_i64();
-                                let s = sdt.scalar(result)?;
+                                let s = scalar_value(dt, result)?;
                                 ret = Some(Ok(ColumnarValue::Scalar(s)));
                                 break;
                             }
@@ -568,3 +530,16 @@ where
     // first map is the iterator, second is for the `Option<_>`
     array.iter().map(|x| x.map(&op).transpose()).collect()
 }
+
+fn scalar_value(dt: &DataType, r: Option<i64>) -> Result<ScalarValue> {
+    match dt {
+        DataType::Date32 => Ok(ScalarValue::Date32(r.and_then(|v| v.to_i32()))),
+        DataType::Timestamp(u, tz) => match u {
+            TimeUnit::Second => Ok(ScalarValue::TimestampSecond(r, tz.clone())),
+            TimeUnit::Millisecond => Ok(ScalarValue::TimestampMillisecond(r, tz.clone())),
+            TimeUnit::Microsecond => Ok(ScalarValue::TimestampMicrosecond(r, tz.clone())),
+            TimeUnit::Nanosecond => Ok(ScalarValue::TimestampNanosecond(r, tz.clone())),
+        },
+        t => Err(internal_datafusion_err!("Unsupported data type: {t:?}")),
+    }
+}
diff --git a/datafusion/functions/src/datetime/to_date.rs b/datafusion/functions/src/datetime/to_date.rs
index e65b5e9599cab..60c6fdf2df975 100644
--- a/datafusion/functions/src/datetime/to_date.rs
+++ b/datafusion/functions/src/datetime/to_date.rs
@@ -84,7 +84,7 @@ impl ToDateFunc {
 
     fn to_date(&self, args: &[ColumnarValue]) -> Result<ColumnarValue> {
         match args.len() {
-            1 => handle::<Date32Type, _, i32>(
+            1 => handle::<Date32Type, _>(
                 args,
                 |s| match Date32Type::parse(s) {
                     Some(v) => Ok(v),
@@ -94,9 +94,9 @@ impl ToDateFunc {
                     )),
                 },
                 "to_date",
-                &ScalarDataType::new(Date32),
+                &Date32,
             ),
-            2.. => handle_multiple::<Date32Type, _, _, i32>(
+            2.. => handle_multiple::<Date32Type, _, _>(
                 args,
                 |s, format| {
                     string_to_timestamp_millis_formatted(s, format)
@@ -109,7 +109,7 @@ impl ToDateFunc {
                 },
                 |n| n,
                 "to_date",
-                &ScalarDataType::new(Date32),
+                &Date32,
             ),
             0 => exec_err!("Unsupported 0 argument count for function to_date"),
         }
diff --git a/datafusion/functions/src/datetime/to_timestamp.rs b/datafusion/functions/src/datetime/to_timestamp.rs
index 728413142304b..9e9524a6c9a3f 100644
--- a/datafusion/functions/src/datetime/to_timestamp.rs
+++ b/datafusion/functions/src/datetime/to_timestamp.rs
@@ -798,20 +798,20 @@ fn to_timestamp_impl<T: ArrowTimestampType + ScalarType<i64>>(
     };
 
     match args.len() {
-        1 => handle::<T, _, i64>(
+        1 => handle::<T, _>(
             args,
             move |s| string_to_timestamp_nanos_with_timezone(&tz, s).map(|n| n / factor),
             name,
-            &ScalarDataType::new(Timestamp(T::UNIT, timezone.clone())),
+            &Timestamp(T::UNIT, timezone.clone()),
         ),
-        n if n >= 2 => handle_multiple::<T, _, _, i64>(
+        n if n >= 2 => handle_multiple::<T, _, _>(
             args,
             move |s, format| {
                 string_to_timestamp_nanos_formatted_with_timezone(&tz, s, format)
             },
             |n| n / factor,
             name,
-            &ScalarDataType::new(Timestamp(T::UNIT, timezone.clone())),
+            &Timestamp(T::UNIT, timezone.clone()),
         ),
         _ => exec_err!("Unsupported 0 argument count for function {name}"),
     }

From 1579ec3b85963725b7d8e8415393181fd42e905c Mon Sep 17 00:00:00 2001
From: kosiew <kosiew@gmail.com>
Date: Mon, 22 Dec 2025 22:53:58 +0800
Subject: [PATCH 10/11] Merge pull request #7
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* refactor: simplify ToTimestamp* constructors using macros for consist…
---
 .../functions/src/datetime/to_timestamp.rs    | 201 +++++-------------
 1 file changed, 48 insertions(+), 153 deletions(-)

diff --git a/datafusion/functions/src/datetime/to_timestamp.rs b/datafusion/functions/src/datetime/to_timestamp.rs
index 9e9524a6c9a3f..1a3bc1ff33f1a 100644
--- a/datafusion/functions/src/datetime/to_timestamp.rs
+++ b/datafusion/functions/src/datetime/to_timestamp.rs
@@ -284,150 +284,46 @@ pub struct ToTimestampNanosFunc {
     timezone: Option<Arc<str>>,
 }
 
-impl Default for ToTimestampFunc {
-    fn default() -> Self {
-        Self::new_with_config(&ConfigOptions::default())
-    }
-}
-
-impl ToTimestampFunc {
-    #[deprecated(since = "52.0.0", note = "use `new_with_config` instead")]
-    /// Deprecated constructor retained for backwards compatibility.
-    ///
-    /// Prefer [`ToTimestampFunc::new_with_config`] which allows specifying the
-    /// timezone via [`ConfigOptions`]. This helper now mirrors the
-    /// canonical default offset (None) provided by `ConfigOptions::default()`.
-    pub fn new() -> Self {
-        Self::new_with_config(&ConfigOptions::default())
-    }
-
-    pub fn new_with_config(config: &ConfigOptions) -> Self {
-        Self {
-            signature: Signature::variadic_any(Volatility::Immutable),
-            timezone: config
-                .execution
-                .time_zone
-                .as_ref()
-                .map(|tz| Arc::from(tz.as_str())),
-        }
-    }
-}
-
-impl Default for ToTimestampSecondsFunc {
-    fn default() -> Self {
-        Self::new_with_config(&ConfigOptions::default())
-    }
-}
-
-impl ToTimestampSecondsFunc {
-    #[deprecated(since = "52.0.0", note = "use `new_with_config` instead")]
-    /// Deprecated constructor retained for backwards compatibility.
-    ///
-    /// Prefer [`ToTimestampSecondsFunc::new_with_config`] which allows specifying the
-    /// timezone via [`ConfigOptions`]. This helper now mirrors the
-    /// canonical default offset (None) provided by `ConfigOptions::default()`.
-    pub fn new() -> Self {
-        Self::new_with_config(&ConfigOptions::default())
-    }
-
-    pub fn new_with_config(config: &ConfigOptions) -> Self {
-        Self {
-            signature: Signature::variadic_any(Volatility::Immutable),
-            timezone: config
-                .execution
-                .time_zone
-                .as_ref()
-                .map(|tz| Arc::from(tz.as_str())),
-        }
-    }
-}
-
-impl Default for ToTimestampMillisFunc {
-    fn default() -> Self {
-        Self::new_with_config(&ConfigOptions::default())
-    }
-}
-
-impl ToTimestampMillisFunc {
-    #[deprecated(since = "52.0.0", note = "use `new_with_config` instead")]
-    /// Deprecated constructor retained for backwards compatibility.
-    ///
-    /// Prefer [`ToTimestampMillisFunc::new_with_config`] which allows specifying the
-    /// timezone via [`ConfigOptions`]. This helper now mirrors the
-    /// canonical default offset (None) provided by `ConfigOptions::default()`.
-    pub fn new() -> Self {
-        Self::new_with_config(&ConfigOptions::default())
-    }
-
-    pub fn new_with_config(config: &ConfigOptions) -> Self {
-        Self {
-            signature: Signature::variadic_any(Volatility::Immutable),
-            timezone: config
-                .execution
-                .time_zone
-                .as_ref()
-                .map(|tz| Arc::from(tz.as_str())),
+/// Macro to generate boilerplate constructors and config methods for ToTimestamp* functions.
+/// Generates: Default impl, deprecated new(), new_with_config(), and extracts timezone from ConfigOptions.
+macro_rules! impl_to_timestamp_constructors {
+    ($func:ty) => {
+        impl Default for $func {
+            fn default() -> Self {
+                Self::new_with_config(&ConfigOptions::default())
+            }
         }
-    }
-}
 
-impl Default for ToTimestampMicrosFunc {
-    fn default() -> Self {
-        Self::new_with_config(&ConfigOptions::default())
-    }
-}
-
-impl ToTimestampMicrosFunc {
-    #[deprecated(since = "52.0.0", note = "use `new_with_config` instead")]
-    /// Deprecated constructor retained for backwards compatibility.
-    ///
-    /// Prefer [`ToTimestampMicrosFunc::new_with_config`] which allows specifying the
-    /// timezone via [`ConfigOptions`]. This helper now mirrors the
-    /// canonical default offset (None) provided by `ConfigOptions::default()`.
-    pub fn new() -> Self {
-        Self::new_with_config(&ConfigOptions::default())
-    }
+        impl $func {
+            #[deprecated(since = "52.0.0", note = "use `new_with_config` instead")]
+            /// Deprecated constructor retained for backwards compatibility.
+            ///
+            /// Prefer [`new_with_config`] which allows specifying the
+            /// timezone via [`ConfigOptions`]. This helper now mirrors the
+            /// canonical default offset (None) provided by `ConfigOptions::default()`.
+            pub fn new() -> Self {
+                Self::new_with_config(&ConfigOptions::default())
+            }
 
-    pub fn new_with_config(config: &ConfigOptions) -> Self {
-        Self {
-            signature: Signature::variadic_any(Volatility::Immutable),
-            timezone: config
-                .execution
-                .time_zone
-                .as_ref()
-                .map(|tz| Arc::from(tz.as_str())),
+            pub fn new_with_config(config: &ConfigOptions) -> Self {
+                Self {
+                    signature: Signature::variadic_any(Volatility::Immutable),
+                    timezone: config
+                        .execution
+                        .time_zone
+                        .as_ref()
+                        .map(|tz| Arc::from(tz.as_str())),
+                }
+            }
         }
-    }
-}
-
-impl Default for ToTimestampNanosFunc {
-    fn default() -> Self {
-        Self::new_with_config(&ConfigOptions::default())
-    }
+    };
 }
 
-impl ToTimestampNanosFunc {
-    #[deprecated(since = "52.0.0", note = "use `new_with_config` instead")]
-    /// Deprecated constructor retained for backwards compatibility.
-    ///
-    /// Prefer [`ToTimestampNanosFunc::new_with_config`] which allows specifying the
-    /// timezone via [`ConfigOptions`]. This helper now mirrors the
-    /// canonical default offset (None) provided by `ConfigOptions::default()`.
-    pub fn new() -> Self {
-        Self::new_with_config(&ConfigOptions::default())
-    }
-
-    pub fn new_with_config(config: &ConfigOptions) -> Self {
-        Self {
-            signature: Signature::variadic_any(Volatility::Immutable),
-            timezone: config
-                .execution
-                .time_zone
-                .as_ref()
-                .map(|tz| Arc::from(tz.as_str())),
-        }
-    }
-}
+impl_to_timestamp_constructors!(ToTimestampFunc);
+impl_to_timestamp_constructors!(ToTimestampSecondsFunc);
+impl_to_timestamp_constructors!(ToTimestampMillisFunc);
+impl_to_timestamp_constructors!(ToTimestampMicrosFunc);
+impl_to_timestamp_constructors!(ToTimestampNanosFunc);
 
 /// to_timestamp SQL function
 ///
@@ -435,6 +331,15 @@ impl ToTimestampNanosFunc {
 /// The supported range for integer input is between `-9223372037` and `9223372036`.
 /// Supported range for string input is between `1677-09-21T00:12:44.0` and `2262-04-11T23:47:16.0`.
 /// Please use `to_timestamp_seconds` for the input outside of supported bounds.
+/// Macro to generate the with_updated_config method for ToTimestamp* functions.
+macro_rules! impl_with_updated_config {
+    () => {
+        fn with_updated_config(&self, config: &ConfigOptions) -> Option<ScalarUDF> {
+            Some(Self::new_with_config(config).into())
+        }
+    };
+}
+
 impl ScalarUDFImpl for ToTimestampFunc {
     fn as_any(&self) -> &dyn Any {
         self
@@ -452,9 +357,7 @@ impl ScalarUDFImpl for ToTimestampFunc {
         Ok(Timestamp(Nanosecond, self.timezone.clone()))
     }
 
-    fn with_updated_config(&self, config: &ConfigOptions) -> Option<ScalarUDF> {
-        Some(Self::new_with_config(config).into())
-    }
+    impl_with_updated_config!();
 
     fn invoke_with_args(
         &self,
@@ -547,9 +450,7 @@ impl ScalarUDFImpl for ToTimestampSecondsFunc {
         Ok(Timestamp(Second, self.timezone.clone()))
     }
 
-    fn with_updated_config(&self, config: &ConfigOptions) -> Option<ScalarUDF> {
-        Some(Self::new_with_config(config).into())
-    }
+    impl_with_updated_config!();
 
     fn invoke_with_args(
         &self,
@@ -611,9 +512,7 @@ impl ScalarUDFImpl for ToTimestampMillisFunc {
         Ok(Timestamp(Millisecond, self.timezone.clone()))
     }
 
-    fn with_updated_config(&self, config: &ConfigOptions) -> Option<ScalarUDF> {
-        Some(Self::new_with_config(config).into())
-    }
+    impl_with_updated_config!();
 
     fn invoke_with_args(
         &self,
@@ -673,9 +572,7 @@ impl ScalarUDFImpl for ToTimestampMicrosFunc {
         Ok(Timestamp(Microsecond, self.timezone.clone()))
     }
 
-    fn with_updated_config(&self, config: &ConfigOptions) -> Option<ScalarUDF> {
-        Some(Self::new_with_config(config).into())
-    }
+    impl_with_updated_config!();
 
     fn invoke_with_args(
         &self,
@@ -735,9 +632,7 @@ impl ScalarUDFImpl for ToTimestampNanosFunc {
         Ok(Timestamp(Nanosecond, self.timezone.clone()))
     }
 
-    fn with_updated_config(&self, config: &ConfigOptions) -> Option<ScalarUDF> {
-        Some(Self::new_with_config(config).into())
-    }
+    impl_with_updated_config!();
 
     fn invoke_with_args(
         &self,

From 783b6846c2f4f8906e04d4235d6d71ad4adcbc99 Mon Sep 17 00:00:00 2001
From: Bruce Ritchie <bruce.ritchie@gmail.com>
Date: Mon, 22 Dec 2025 10:09:36 -0500
Subject: [PATCH 11/11] Small doc fix.

---
 datafusion/functions/src/datetime/to_timestamp.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/datafusion/functions/src/datetime/to_timestamp.rs b/datafusion/functions/src/datetime/to_timestamp.rs
index 1a3bc1ff33f1a..58077694b07a0 100644
--- a/datafusion/functions/src/datetime/to_timestamp.rs
+++ b/datafusion/functions/src/datetime/to_timestamp.rs
@@ -298,7 +298,7 @@ macro_rules! impl_to_timestamp_constructors {
             #[deprecated(since = "52.0.0", note = "use `new_with_config` instead")]
             /// Deprecated constructor retained for backwards compatibility.
             ///
-            /// Prefer [`new_with_config`] which allows specifying the
+            /// Prefer `new_with_config` which allows specifying the
             /// timezone via [`ConfigOptions`]. This helper now mirrors the
             /// canonical default offset (None) provided by `ConfigOptions::default()`.
             pub fn new() -> Self {
