Allow logical optimizer to be run without evaluating now()

Author: adriangb

datafusion/core/src/execution/session_state.rs

@@ -2063,7 +2063,7 @@ impl datafusion_execution::TaskContextProvider for SessionState {
 }
 
 impl OptimizerConfig for SessionState {
-    fn query_execution_start_time(&self) -> DateTime<Utc> {
+    fn query_execution_start_time(&self) -> Option<DateTime<Utc>> {
         self.execution_props.query_execution_start_time
     }
 
@@ -2135,13 +2135,17 @@ impl SimplifyInfo for SessionSimplifyProvider<'_> {
         expr.nullable(self.df_schema)
     }
 
-    fn execution_props(&self) -> &ExecutionProps {
-        self.state.execution_props()
-    }
-
     fn get_data_type(&self, expr: &Expr) -> datafusion_common::Result<DataType> {
         expr.get_type(self.df_schema)
     }
+
+    fn query_execution_start_time(&self) -> Option<DateTime<Utc>> {
+        self.state.execution_props().query_execution_start_time
+    }
+
+    fn config_options(&self) -> Option<&Arc<ConfigOptions>> {
+        self.state.execution_props().config_options.as_ref()
+    }
 }
 
 #[derive(Debug)]

datafusion/core/tests/expr_api/simplification.rs

@@ -26,6 +26,7 @@ use chrono::{DateTime, TimeZone, Utc};
 use datafusion::{error::Result, execution::context::ExecutionProps, prelude::*};
 use datafusion_common::ScalarValue;
 use datafusion_common::cast::as_int32_array;
+use datafusion_common::config::ConfigOptions;
 use datafusion_common::{DFSchemaRef, ToDFSchema};
 use datafusion_expr::expr::ScalarFunction;
 use datafusion_expr::logical_plan::builder::table_scan_with_filters;
@@ -66,13 +67,17 @@ impl SimplifyInfo for MyInfo {
         expr.nullable(self.schema.as_ref())
     }
 
-    fn execution_props(&self) -> &ExecutionProps {
-        &self.execution_props
-    }
-
     fn get_data_type(&self, expr: &Expr) -> Result<DataType> {
         expr.get_type(self.schema.as_ref())
     }
+
+    fn query_execution_start_time(&self) -> Option<DateTime<Utc>> {
+        self.execution_props.query_execution_start_time
+    }
+
+    fn config_options(&self) -> Option<&Arc<ConfigOptions>> {
+        self.execution_props.config_options.as_ref()
+    }
 }
 
 impl From<DFSchemaRef> for MyInfo {
@@ -523,6 +528,51 @@ fn multiple_now() -> Result<()> {
     Ok(())
 }
 
+/// Test that `now()` is simplified to a literal when execution start time is set,
+/// but remains as an expression when no execution start time is available.
+#[test]
+fn now_simplification_with_and_without_start_time() {
+    let schema = expr_test_schema();
+
+    // Case 1: With execution start time set, now() should be simplified to a literal
+    {
+        let start_time = Utc::now();
+        let info = MyInfo {
+            schema: Arc::clone(&schema),
+            execution_props: ExecutionProps::new()
+                .with_query_execution_start_time(start_time),
+        };
+        let simplifier = ExprSimplifier::new(info);
+        let simplified = simplifier.simplify(now()).expect("simplify should succeed");
+
+        // Should be a literal timestamp
+        match simplified {
+            Expr::Literal(ScalarValue::TimestampNanosecond(Some(ts), _), _) => {
+                assert_eq!(ts, start_time.timestamp_nanos_opt().unwrap());
+            }
+            other => panic!("Expected timestamp literal, got: {other:?}"),
+        }
+    }
+
+    // Case 2: Without execution start time, now() should remain as a function call
+    {
+        let info = MyInfo {
+            schema: Arc::clone(&schema),
+            execution_props: ExecutionProps::new(), // No start time set
+        };
+        let simplifier = ExprSimplifier::new(info);
+        let simplified = simplifier.simplify(now()).expect("simplify should succeed");
+
+        // Should remain as a scalar function (not simplified)
+        match simplified {
+            Expr::ScalarFunction(func) => {
+                assert_eq!(func.name(), "now");
+            }
+            other => panic!("Expected now() to remain unsimplified, got: {other:?}"),
+        }
+    }
+}
+
 // ------------------------------
 // --- Simplifier tests -----
 // ------------------------------
@@ -566,7 +616,8 @@ fn test_simplify_with_cycle_count(
 ) {
     let info: MyInfo = MyInfo {
         schema: expr_test_schema(),
-        execution_props: ExecutionProps::new(),
+        execution_props: ExecutionProps::new()
+            .with_query_execution_start_time(Utc::now()),
     };
     let simplifier = ExprSimplifier::new(info);
     let (simplified_expr, count) = simplifier

datafusion/expr/src/execution_props.rs

@@ -16,7 +16,7 @@
 // under the License.
 
 use crate::var_provider::{VarProvider, VarType};
-use chrono::{DateTime, TimeZone, Utc};
+use chrono::{DateTime, Utc};
 use datafusion_common::HashMap;
 use datafusion_common::alias::AliasGenerator;
 use datafusion_common::config::ConfigOptions;
@@ -33,7 +33,9 @@ use std::sync::Arc;
 /// done so during predicate pruning and expression simplification
 #[derive(Clone, Debug)]
 pub struct ExecutionProps {
-    pub query_execution_start_time: DateTime<Utc>,
+    /// The time at which the query execution started. If `None`,
+    /// functions like `now()` will not be simplified during optimization.
+    pub query_execution_start_time: Option<DateTime<Utc>>,
     /// Alias generator used by subquery optimizer rules
     pub alias_generator: Arc<AliasGenerator>,
     /// Snapshot of config options when the query started
@@ -52,9 +54,7 @@ impl ExecutionProps {
     /// Creates a new execution props
     pub fn new() -> Self {
         ExecutionProps {
-            // Set this to a fixed sentinel to make it obvious if this is
-            // not being updated / propagated correctly
-            query_execution_start_time: Utc.timestamp_nanos(0),
+            query_execution_start_time: None,
             alias_generator: Arc::new(AliasGenerator::new()),
             config_options: None,
             var_providers: None,
@@ -66,7 +66,7 @@ impl ExecutionProps {
         mut self,
         query_execution_start_time: DateTime<Utc>,
     ) -> Self {
-        self.query_execution_start_time = query_execution_start_time;
+        self.query_execution_start_time = Some(query_execution_start_time);
         self
     }
 
@@ -79,7 +79,7 @@ impl ExecutionProps {
     /// Marks the execution of query started timestamp.
     /// This also instantiates a new alias generator.
     pub fn mark_start_execution(&mut self, config_options: Arc<ConfigOptions>) -> &Self {
-        self.query_execution_start_time = Utc::now();
+        self.query_execution_start_time = Some(Utc::now());
         self.alias_generator = Arc::new(AliasGenerator::new());
         self.config_options = Some(config_options);
         &*self
@@ -126,7 +126,7 @@ mod test {
     fn debug() {
         let props = ExecutionProps::new();
         assert_eq!(
-            "ExecutionProps { query_execution_start_time: 1970-01-01T00:00:00Z, alias_generator: AliasGenerator { next_id: 1 }, config_options: None, var_providers: None }",
+            "ExecutionProps { query_execution_start_time: None, alias_generator: AliasGenerator { next_id: 1 }, config_options: None, var_providers: None }",
             format!("{props:?}")
         );
     }

datafusion/expr/src/simplify.rs

@@ -17,7 +17,11 @@
 
 //! Structs and traits to provide the information needed for expression simplification.
 
+use std::sync::Arc;
+
 use arrow::datatypes::DataType;
+use chrono::{DateTime, Utc};
+use datafusion_common::config::ConfigOptions;
 use datafusion_common::{DFSchemaRef, Result, internal_datafusion_err};
 
 use crate::{Expr, ExprSchemable, execution_props::ExecutionProps};
@@ -35,43 +39,85 @@ pub trait SimplifyInfo {
     /// Returns true of this expr is nullable (could possibly be NULL)
     fn nullable(&self, expr: &Expr) -> Result<bool>;
 
-    /// Returns details needed for partial expression evaluation
-    fn execution_props(&self) -> &ExecutionProps;
-
     /// Returns data type of this expr needed for determining optimized int type of a value
     fn get_data_type(&self, expr: &Expr) -> Result<DataType>;
+
+    /// Returns the time at which the query execution started.
+    /// If `None`, time-dependent functions like `now()` will not be simplified.
+    fn query_execution_start_time(&self) -> Option<DateTime<Utc>>;
+
+    /// Returns the configuration options for the session.
+    fn config_options(&self) -> Option<&Arc<ConfigOptions>>;
 }
 
-/// Provides simplification information based on DFSchema and
-/// [`ExecutionProps`]. This is the default implementation used by DataFusion
+/// Provides simplification information based on schema, query execution time,
+/// and configuration options.
 ///
 /// # Example
 /// See the `simplify_demo` in the [`expr_api` example]
 ///
 /// [`expr_api` example]: https://github.com/apache/datafusion/blob/main/datafusion-examples/examples/query_planning/expr_api.rs
 #[derive(Debug, Clone)]
-pub struct SimplifyContext<'a> {
+pub struct SimplifyContext {
     schema: Option<DFSchemaRef>,
-    props: &'a ExecutionProps,
+    query_execution_start_time: Option<DateTime<Utc>>,
+    config_options: Option<Arc<ConfigOptions>>,
 }
 
-impl<'a> SimplifyContext<'a> {
-    /// Create a new SimplifyContext
-    pub fn new(props: &'a ExecutionProps) -> Self {
+impl SimplifyContext {
+    /// Create a new SimplifyContext from [`ExecutionProps`].
+    ///
+    /// This constructor extracts `query_execution_start_time` and `config_options`
+    /// from the provided `ExecutionProps`.
+    pub fn new(props: &ExecutionProps) -> Self {
+        Self {
+            schema: None,
+            query_execution_start_time: props.query_execution_start_time,
+            config_options: props.config_options.clone(),
+        }
+    }
+
+    /// Create a new empty SimplifyContext.
+    ///
+    /// This is useful when you don't need time-dependent simplification
+    /// (like `now()`) or config-based simplification.
+    pub fn new_empty() -> Self {
         Self {
             schema: None,
-            props,
+            query_execution_start_time: None,
+            config_options: None,
         }
     }
 
+    /// Set the query execution start time
+    pub fn with_query_execution_start_time(
+        mut self,
+        query_execution_start_time: Option<DateTime<Utc>>,
+    ) -> Self {
+        self.query_execution_start_time = query_execution_start_time;
+        self
+    }
+
+    /// Set the configuration options
+    pub fn with_config_options(mut self, config_options: Arc<ConfigOptions>) -> Self {
+        self.config_options = Some(config_options);
+        self
+    }
+
     /// Register a [`DFSchemaRef`] with this context
     pub fn with_schema(mut self, schema: DFSchemaRef) -> Self {
         self.schema = Some(schema);
         self
     }
 }
 
-impl SimplifyInfo for SimplifyContext<'_> {
+impl Default for SimplifyContext {
+    fn default() -> Self {
+        Self::new_empty()
+    }
+}
+
+impl SimplifyInfo for SimplifyContext {
     /// Returns true if this Expr has boolean type
     fn is_boolean_type(&self, expr: &Expr) -> Result<bool> {
         if let Some(schema) = &self.schema
@@ -99,8 +145,12 @@ impl SimplifyInfo for SimplifyContext<'_> {
         expr.get_type(schema)
     }
 
-    fn execution_props(&self) -> &ExecutionProps {
-        self.props
+    fn query_execution_start_time(&self) -> Option<DateTime<Utc>> {
+        self.query_execution_start_time
+    }
+
+    fn config_options(&self) -> Option<&Arc<ConfigOptions>> {
+        self.config_options.as_ref()
     }
 }
 

datafusion/functions/src/datetime/current_date.rs

@@ -99,14 +99,15 @@ impl ScalarUDFImpl for CurrentDateFunc {
 
     fn simplify(
         &self,
-        _args: Vec<Expr>,
+        args: Vec<Expr>,
         info: &dyn SimplifyInfo,
     ) -> Result<ExprSimplifyResult> {
-        let now_ts = info.execution_props().query_execution_start_time;
+        let Some(now_ts) = info.query_execution_start_time() else {
+            return Ok(ExprSimplifyResult::Original(args));
+        };
 
         // Get timezone from config and convert to local time
         let days = info
-            .execution_props()
             .config_options()
             .and_then(|config| {
                 config

datafusion/functions/src/datetime/current_time.rs

@@ -95,14 +95,15 @@ impl ScalarUDFImpl for CurrentTimeFunc {
 
     fn simplify(
         &self,
-        _args: Vec<Expr>,
+        args: Vec<Expr>,
         info: &dyn SimplifyInfo,
     ) -> Result<ExprSimplifyResult> {
-        let now_ts = info.execution_props().query_execution_start_time;
+        let Some(now_ts) = info.query_execution_start_time() else {
+            return Ok(ExprSimplifyResult::Original(args));
+        };
 
         // Try to get timezone from config and convert to local time
         let nano = info
-            .execution_props()
             .config_options()
             .and_then(|config| {
                 config
@@ -145,6 +146,7 @@ mod tests {
     use super::*;
     use arrow::datatypes::{DataType, TimeUnit::Nanosecond};
     use chrono::{DateTime, Utc};
+    use datafusion_common::config::ConfigOptions;
     use datafusion_common::{Result, ScalarValue};
     use datafusion_expr::execution_props::ExecutionProps;
     use datafusion_expr::simplify::{ExprSimplifyResult, SimplifyInfo};
@@ -163,17 +165,21 @@ mod tests {
             Ok(true)
         }
 
-        fn execution_props(&self) -> &ExecutionProps {
-            &self.execution_props
-        }
-
         fn get_data_type(&self, _expr: &Expr) -> Result<DataType> {
             Ok(Time64(Nanosecond))
         }
+
+        fn query_execution_start_time(&self) -> Option<DateTime<Utc>> {
+            self.execution_props.query_execution_start_time
+        }
+
+        fn config_options(&self) -> Option<&Arc<ConfigOptions>> {
+            self.execution_props.config_options.as_ref()
+        }
     }
 
     fn set_session_timezone_env(tz: &str, start_time: DateTime<Utc>) -> MockSimplifyInfo {
-        let mut config = datafusion_common::config::ConfigOptions::default();
+        let mut config = ConfigOptions::default();
         config.execution.time_zone = if tz.is_empty() {
             None
         } else {