[task #5568] add_to_unixtime_function

Signed-off-by: tangruilin <[email protected]>

Author: Tangruilin

datafusion/expr/src/built_in_function.rs

@@ -277,6 +277,8 @@ pub enum BuiltinScalarFunction {
     FromUnixtime,
     /// to_date
     ToDate,
+    /// to_unixtime
+    ToUnixtime,
     ///now
     Now,
     ///current_date
@@ -475,6 +477,7 @@ impl BuiltinScalarFunction {
             BuiltinScalarFunction::Struct => Volatility::Immutable,
             BuiltinScalarFunction::FromUnixtime => Volatility::Immutable,
             BuiltinScalarFunction::ToDate => Volatility::Immutable,
+            BuiltinScalarFunction::ToUnixtime => Volatility::Immutable,
             BuiltinScalarFunction::ArrowTypeof => Volatility::Immutable,
             BuiltinScalarFunction::OverLay => Volatility::Immutable,
             BuiltinScalarFunction::Levenshtein => Volatility::Immutable,
@@ -785,6 +788,7 @@ impl BuiltinScalarFunction {
             BuiltinScalarFunction::ToTimestampSeconds => Ok(Timestamp(Second, None)),
             BuiltinScalarFunction::FromUnixtime => Ok(Timestamp(Second, None)),
             BuiltinScalarFunction::ToDate => Ok(Date32),
+            BuiltinScalarFunction::ToUnixtime => Ok(Int64),
             BuiltinScalarFunction::Now => {
                 Ok(Timestamp(Nanosecond, Some("+00:00".into())))
             }
@@ -1063,6 +1067,9 @@ impl BuiltinScalarFunction {
                 Signature::uniform(1, vec![Int64], self.volatility())
             }
             BuiltinScalarFunction::ToDate => Signature::variadic_any(self.volatility()),
+            BuiltinScalarFunction::ToUnixtime => {
+                Signature::variadic_any(self.volatility())
+            }
             BuiltinScalarFunction::Digest => Signature::one_of(
                 vec![
                     Exact(vec![Utf8, Utf8]),
@@ -1496,6 +1503,7 @@ impl BuiltinScalarFunction {
             BuiltinScalarFunction::ToTimestampNanos => &["to_timestamp_nanos"],
             BuiltinScalarFunction::FromUnixtime => &["from_unixtime"],
             BuiltinScalarFunction::ToDate => &["to_date"],
+            BuiltinScalarFunction::ToUnixtime => &["to_unixtime"],
 
             // hashing functions
             BuiltinScalarFunction::Digest => &["digest"],

datafusion/physical-expr/src/datetime_expressions.rs

@@ -58,6 +58,8 @@ use datafusion_common::{
     ScalarValue,
 };
 use datafusion_expr::ColumnarValue;
+use petgraph::visit::Data;
+use uuid::timestamp;
 
 use crate::expressions::cast_column;
 
@@ -1878,6 +1880,30 @@ pub fn from_unixtime_invoke(args: &[ColumnarValue]) -> Result<ColumnarValue> {
     }
 }
 
+/// to_unixtime() SQL function implementation
+pub fn to_unixtime_invoke(args: &[ColumnarValue]) -> Result<ColumnarValue> {
+    if args.len() != 1 {
+        return exec_err!(
+            "to_unixtime function require 1 argument, get {}",
+            args.len()
+        );
+    }
+
+    match args[0].data_type() {
+        DataType::Null | DataType::Int32 | DataType::Int64 => Ok(args[0].clone()),
+        DataType::Timestamp(_, None) => cast_column(&args[0], &DataType::Int64, None),
+        DataType::Utf8 => {
+            cast_column(&to_timestamp_seconds(args)?, &DataType::Int64, None)
+        }
+        other => {
+            exec_err!(
+                "Unsupported data type {:?} for function to_unixtime function",
+                other
+            )
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;

datafusion/physical-expr/src/functions.rs

@@ -523,6 +523,9 @@ pub fn create_physical_fun(
         BuiltinScalarFunction::FromUnixtime => {
             Arc::new(datetime_expressions::from_unixtime_invoke)
         }
+        BuiltinScalarFunction::ToUnixtime => {
+            Arc::new(datetime_expressions::to_unixtime_invoke)
+        }
         BuiltinScalarFunction::ToDate => Arc::new(datetime_expressions::to_date_invoke),
         BuiltinScalarFunction::InitCap => Arc::new(|args| match args[0].data_type() {
             DataType::Utf8 => {

datafusion/proto/proto/datafusion.proto

@@ -1,5 +1,6 @@
 /*
 
+
  * 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
@@ -684,6 +685,7 @@ enum ScalarFunction {
   RegexpLike = 135;
   ToChar = 136;
   ToDate = 137;
+  ToUnixtime = 138;
 }
 
 message ScalarFunctionNode {

datafusion/proto/src/generated/pbjson.rs

@@ -571,6 +571,7 @@ impl From<&protobuf::ScalarFunction> for BuiltinScalarFunction {
             ScalarFunction::SubstrIndex => Self::SubstrIndex,
             ScalarFunction::FindInSet => Self::FindInSet,
             ScalarFunction::ToDate => Self::ToDate,
+            ScalarFunction::ToUnixtime => Self::ToUnixtime,
         }
     }
 }
@@ -1821,6 +1822,16 @@ pub fn parse_expr(
                         args,
                     )))
                 }
+                ScalarFunction::ToUnixtime => {
+                    let args: Vec<_> = args
+                        .iter()
+                        .map(|expr| parse_expr(expr, registry))
+                        .collect::<std::result::Result<_, _>>()?;
+                    Ok(Expr::ScalarFunction(expr::ScalarFunction::new(
+                        BuiltinScalarFunction::ToUnixtime,
+                        args,
+                    )))
+                }
             }
         }
         ExprType::ScalarUdfExpr(protobuf::ScalarUdfExprNode { fun_name, args }) => {

datafusion/proto/src/generated/prost.rs

@@ -1550,6 +1550,7 @@ impl TryFrom<&BuiltinScalarFunction> for protobuf::ScalarFunction {
             BuiltinScalarFunction::SubstrIndex => Self::SubstrIndex,
             BuiltinScalarFunction::FindInSet => Self::FindInSet,
             BuiltinScalarFunction::ToDate => Self::ToDate,
+            BuiltinScalarFunction::ToUnixtime => Self::ToUnixtime,
         };
 
         Ok(scalar_function)