Implement current_date scalar function (apache#4022)

* current_date impl

* Update datafusion/expr/src/expr_fn.rs

Co-authored-by: Andrew Lamb <[email protected]>

* Update datafusion/physical-expr/src/functions.rs

Co-authored-by: Andrew Lamb <[email protected]>

Co-authored-by: Andrew Lamb <[email protected]>

Author: 2 people

datafusion/core/src/physical_plan/functions.rs

@@ -211,6 +211,7 @@ pub fn return_type(
         BuiltinScalarFunction::UtcTimestamp => {
             Ok(DataType::Timestamp(TimeUnit::Nanosecond, None))
         }
+        BuiltinScalarFunction::CurrentDate => Ok(DataType::Date32),
         BuiltinScalarFunction::Translate => {
             utf8_to_str_type(&input_expr_types[0], "translate")
         }
@@ -893,6 +894,12 @@ pub fn create_physical_fun(
                 execution_props.query_execution_start_time,
             ))
         }
+        BuiltinScalarFunction::CurrentDate => {
+            // bind value for current_date at plan time
+            Arc::new(datetime_expressions::make_current_date(
+                execution_props.query_execution_start_time,
+            ))
+        }
         BuiltinScalarFunction::InitCap => Arc::new(|args| match args[0].data_type() {
             DataType::Utf8 => {
                 make_scalar_function(string_expressions::initcap::<i32>)(args)

datafusion/core/tests/sql/timestamp.rs

@@ -814,3 +814,35 @@ async fn group_by_timestamp_millis() -> Result<()> {
     assert_batches_eq!(expected, &actual);
     Ok(())
 }
+
+#[tokio::test]
+async fn test_current_date() -> Result<()> {
+    let ctx = SessionContext::new();
+
+    let sql = "select current_date() dt";
+    let results = execute_to_batches(&ctx, sql).await;
+    assert_eq!(
+        results[0]
+            .schema()
+            .field_with_name("dt")
+            .unwrap()
+            .data_type()
+            .to_owned(),
+        DataType::Date32
+    );
+
+    let sql = "select case when current_date() = cast(now() as date) then 'OK' else 'FAIL' end result";
+    let results = execute_to_batches(&ctx, sql).await;
+
+    let expected = vec![
+        "+--------+",
+        "| result |",
+        "+--------+",
+        "| OK     |",
+        "+--------+",
+    ];
+
+    assert_batches_eq!(expected, &results);
+
+    Ok(())
+}

datafusion/expr/src/built_in_function.rs

@@ -150,7 +150,10 @@ pub enum BuiltinScalarFunction {
     ToTimestampSeconds,
     ///now
     Now,
+    ///utctimestamp
     UtcTimestamp,
+    ///current_date
+    CurrentDate,
     /// translate
     Translate,
     /// trim
@@ -170,6 +173,7 @@ impl BuiltinScalarFunction {
             BuiltinScalarFunction::Random
                 | BuiltinScalarFunction::Now
                 | BuiltinScalarFunction::UtcTimestamp
+                | BuiltinScalarFunction::CurrentDate
         )
     }
     /// Returns the [Volatility] of the builtin function.
@@ -244,6 +248,7 @@ impl BuiltinScalarFunction {
             // Stable builtin functions
             BuiltinScalarFunction::Now => Volatility::Stable,
             BuiltinScalarFunction::UtcTimestamp => Volatility::Stable,
+            BuiltinScalarFunction::CurrentDate => Volatility::Stable,
 
             // Volatile builtin functions
             BuiltinScalarFunction::Random => Volatility::Volatile,
@@ -297,6 +302,7 @@ impl FromStr for BuiltinScalarFunction {
             "concat" => BuiltinScalarFunction::Concat,
             "concat_ws" => BuiltinScalarFunction::ConcatWithSeparator,
             "chr" => BuiltinScalarFunction::Chr,
+            "current_date" => BuiltinScalarFunction::CurrentDate,
             "date_part" | "datepart" => BuiltinScalarFunction::DatePart,
             "date_trunc" | "datetrunc" => BuiltinScalarFunction::DateTrunc,
             "initcap" => BuiltinScalarFunction::InitCap,

datafusion/expr/src/expr_fn.rs

@@ -324,6 +324,14 @@ pub fn coalesce(args: Vec<Expr>) -> Expr {
     }
 }
 
+/// Returns current UTC date as a [`DataType::Date32`] value
+pub fn current_date() -> Expr {
+    Expr::ScalarFunction {
+        fun: built_in_function::BuiltinScalarFunction::CurrentDate,
+        args: vec![],
+    }
+}
+
 #[cfg(test)]
 mod test {
     use super::*;

datafusion/physical-expr/src/datetime_expressions.rs

@@ -356,6 +356,24 @@ pub fn make_now(
     }
 }
 
+/// Create an implementation of `current_date()` that always returns the
+/// specified current date.
+///
+/// The semantics of `current_date()` require it to return the same value
+/// wherever it appears within a single statement. This value is
+/// chosen during planning time.
+pub fn make_current_date(
+    now_ts: DateTime<Utc>,
+) -> impl Fn(&[ColumnarValue]) -> Result<ColumnarValue> {
+    let days = Some(
+        now_ts.num_days_from_ce()
+            - NaiveDate::from_ymd_opt(1970, 1, 1)
+                .unwrap()
+                .num_days_from_ce(),
+    );
+    move |_arg| Ok(ColumnarValue::Scalar(ScalarValue::Date32(days)))
+}
+
 fn quarter_month(date: &NaiveDateTime) -> u32 {
     1 + 3 * ((date.month() - 1) / 3)
 }

datafusion/proto/proto/datafusion.proto

@@ -189,6 +189,8 @@ enum ScalarFunction {
   Trim=61;
   Upper=62;
   Coalesce=63;
+  // Upstream
+  CurrentDate=70;
   // Cubesql
   UtcTimestamp=101;
   ToDayInterval=102;

datafusion/proto/src/from_proto.rs

@@ -428,6 +428,7 @@ impl From<&protobuf::ScalarFunction> for BuiltinScalarFunction {
             ScalarFunction::ToTimestampMicros => Self::ToTimestampMicros,
             ScalarFunction::ToTimestampSeconds => Self::ToTimestampSeconds,
             ScalarFunction::Now => Self::Now,
+            ScalarFunction::CurrentDate => Self::CurrentDate,
             ScalarFunction::Translate => Self::Translate,
             ScalarFunction::RegexpMatch => Self::RegexpMatch,
             ScalarFunction::Coalesce => Self::Coalesce,

datafusion/proto/src/to_proto.rs

@@ -1084,6 +1084,7 @@ impl TryFrom<&BuiltinScalarFunction> for protobuf::ScalarFunction {
             BuiltinScalarFunction::ToTimestampMicros => Self::ToTimestampMicros,
             BuiltinScalarFunction::ToTimestampSeconds => Self::ToTimestampSeconds,
             BuiltinScalarFunction::Now => Self::Now,
+            BuiltinScalarFunction::CurrentDate => Self::CurrentDate,
             BuiltinScalarFunction::Translate => Self::Translate,
             BuiltinScalarFunction::RegexpMatch => Self::RegexpMatch,
             BuiltinScalarFunction::Coalesce => Self::Coalesce,