From b46f0d852480a6f53c970bd5b237f0eac76f5f55 Mon Sep 17 00:00:00 2001
From: tangruilin <tang.ruilin@foxmail.com>
Date: Wed, 31 Jan 2024 09:44:52 +0800
Subject: [PATCH] [task #5568] add_to_unixtime_function

Signed-off-by: tangruilin <tang.ruilin@foxmail.com>
---
 datafusion/expr/src/built_in_function.rs      |  8 ++++++
 .../physical-expr/src/datetime_expressions.rs | 26 +++++++++++++++++++
 datafusion/physical-expr/src/functions.rs     |  3 +++
 datafusion/proto/proto/datafusion.proto       |  2 ++
 4 files changed, 39 insertions(+)

diff --git a/datafusion/expr/src/built_in_function.rs b/datafusion/expr/src/built_in_function.rs
index d10045ff9f602..2dc6b4a31330d 100644
--- a/datafusion/expr/src/built_in_function.rs
+++ b/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"],
diff --git a/datafusion/physical-expr/src/datetime_expressions.rs b/datafusion/physical-expr/src/datetime_expressions.rs
index fd57485555c6e..8c7c57b35621b 100644
--- a/datafusion/physical-expr/src/datetime_expressions.rs
+++ b/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;
diff --git a/datafusion/physical-expr/src/functions.rs b/datafusion/physical-expr/src/functions.rs
index 584901d37d284..2cf2674b7678d 100644
--- a/datafusion/physical-expr/src/functions.rs
+++ b/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 => {
diff --git a/datafusion/proto/proto/datafusion.proto b/datafusion/proto/proto/datafusion.proto
index 9015d9ee9b001..9d6aaf75eddc4 100644
--- a/datafusion/proto/proto/datafusion.proto
+++ b/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 {