feat: add configurable UDF for now() function with timezone support

Author: Weijun-H

datafusion/core/src/execution/context/mod.rs

@@ -81,6 +81,8 @@ use datafusion_expr::{
     planner::ExprPlanner,
     Expr, UserDefinedLogicalNode, WindowUDF,
 };
+use datafusion_functions::datetime::now::NowFunc;
+use datafusion_functions::make_udf_function_with_config;
 use datafusion_optimizer::analyzer::type_coercion::TypeCoercion;
 use datafusion_optimizer::Analyzer;
 use datafusion_optimizer::{AnalyzerRule, OptimizerRule};
@@ -1073,6 +1075,19 @@ impl SessionContext {
             let mut state = self.state.write();
             state.config_mut().options_mut().set(&variable, &value)?;
             drop(state);
+
+            // Register UDFs that return values based on session configuration
+            // e.g. now() which depends on the time_zone configuration option
+            if variable == "datafusion.execution.time_zone" {
+                let state = self.state.read();
+                let config_options = state.config().options().clone();
+                drop(state);
+                let now_udf = {
+                    make_udf_function_with_config!(NowFunc, now, &ConfigOptions);
+                    now(&config_options)
+                };
+                self.state.write().register_udf(now_udf)?;
+            }
         }
 
         self.return_empty_dataframe()

datafusion/functions/src/datetime/now.rs

@@ -19,7 +19,9 @@ use arrow::datatypes::DataType::Timestamp;
 use arrow::datatypes::TimeUnit::Nanosecond;
 use arrow::datatypes::{DataType, Field, FieldRef};
 use std::any::Any;
+use std::sync::Arc;
 
+use datafusion_common::config::ConfigOptions;
 use datafusion_common::{internal_err, Result, ScalarValue};
 use datafusion_expr::simplify::{ExprSimplifyResult, SimplifyInfo};
 use datafusion_expr::{
@@ -41,6 +43,7 @@ The `now()` return value is determined at query time and will return the same ti
 pub struct NowFunc {
     signature: Signature,
     aliases: Vec<String>,
+    timezone: Option<Arc<str>>,
 }
 
 impl Default for NowFunc {
@@ -54,6 +57,15 @@ impl NowFunc {
         Self {
             signature: Signature::nullary(Volatility::Stable),
             aliases: vec!["current_timestamp".to_string()],
+            timezone: Some(Arc::from("+00")),
+        }
+    }
+
+    pub fn new_with_config(config: &ConfigOptions) -> Self {
+        Self {
+            signature: Signature::nullary(Volatility::Stable),
+            aliases: vec!["current_timestamp".to_string()],
+            timezone: Some(Arc::from(config.execution.time_zone.as_str())),
         }
     }
 }
@@ -80,7 +92,7 @@ impl ScalarUDFImpl for NowFunc {
     fn return_field_from_args(&self, _args: ReturnFieldArgs) -> Result<FieldRef> {
         Ok(Field::new(
             self.name(),
-            Timestamp(Nanosecond, Some("+00".into())),
+            Timestamp(Nanosecond, self.timezone.clone()),
             false,
         )
         .into())

datafusion/functions/src/macros.rs

@@ -89,6 +89,22 @@ macro_rules! make_udf_function {
     };
 }
 
+/// Creates a singleton `ScalarUDF` of the `$UDF` function and a function
+/// named `$NAME` which returns that singleton. The function takes a
+/// configuration argument of type `$CONFIG_TYPE` to create the UDF.
+#[macro_export]
+macro_rules! make_udf_function_with_config {
+    ($UDF:ty, $NAME:ident, $CONFIG_TYPE:ty) => {
+        #[allow(rustdoc::redundant_explicit_links)]
+        #[doc = concat!("Return a [`ScalarUDF`](datafusion_expr::ScalarUDF) implementation of ", stringify!($NAME))]
+        pub fn $NAME(config: $CONFIG_TYPE) -> std::sync::Arc<datafusion_expr::ScalarUDF> {
+            std::sync::Arc::new(datafusion_expr::ScalarUDF::new_from_impl(
+                <$UDF>::new_with_config(&config),
+            ))
+        }
+    };
+}
+
 /// Macro creates a sub module if the feature is not enabled
 ///
 /// The rationale for providing stub functions is to help users to configure datafusion