Add function castBIGINT_timestamp (apache#22)

* Add function castBIGINT_timestamp

* fix

* wip

Author: zhztheplayer

cpp/src/arrow/compute/kernels/scalar_cast_test.cc

@@ -1101,6 +1101,17 @@ TEST(Cast, TimestampToTimestamp) {
     options.allow_time_truncate = true;
     CheckCast(will_be_truncated, coarse, options);
   }
+
+  for (auto types : {
+      TimestampTypePair{timestamp(TimeUnit::MILLI, "UTC+8"), timestamp(TimeUnit::MILLI)}
+  }) {
+    auto coarse = ArrayFromJSON(types.coarse, "[0, null, 200000000000, 1000000000, 2000000000]");
+    auto promoted =
+        ArrayFromJSON(types.fine, "[0, null, 200000000000, 1000000000, 2000000000]");
+
+    // multiply/promote
+    CheckCast(coarse, promoted);
+  }
 }
 
 TEST(Cast, TimestampZeroCopy) {

cpp/src/gandiva/function_registry_common.h

@@ -55,6 +55,9 @@ inline DataTypePtr time32() { return arrow::time32(arrow::TimeUnit::MILLI); }
 inline DataTypePtr time64() { return arrow::time64(arrow::TimeUnit::MICRO); }
 
 inline DataTypePtr timestamp() { return arrow::timestamp(arrow::TimeUnit::MILLI); }
+
+inline DataTypePtr timestampusutc() { return arrow::timestamp(arrow::TimeUnit::MICRO, "UTC"); }
+
 inline DataTypePtr decimal128() { return arrow::decimal(38, 0); }
 
 struct KeyHash {
@@ -268,7 +271,7 @@ typedef std::unordered_map<const FunctionSignature*, const NativeFunction*, KeyH
 
 // Iterate the inner macro over all date types
 #define DATE_TYPES(INNER, NAME, ALIASES) \
-  INNER(NAME, ALIASES, date64), INNER(NAME, ALIASES, timestamp)
+  INNER(NAME, ALIASES, date64), INNER(NAME, ALIASES, timestamp), INNER(NAME, ALIASES, timestampusutc)
 
 // Iterate the inner macro over all time types
 #define TIME_TYPES(INNER, NAME, ALIASES) INNER(NAME, ALIASES, time32)

cpp/src/gandiva/function_registry_datetime.cc

@@ -96,6 +96,9 @@ std::vector<NativeFunction> GetDateTimeFunctionRegistry() {
       NativeFunction("castTIME", {}, DataTypeVector{timestamp()}, time32(),
                      kResultNullIfNull, "castTIME_timestamp"),
 
+      NativeFunction("castBIGINT", {}, DataTypeVector{timestamp()}, int64(),
+                     kResultNullIfNull, "castBIGINT_timestamp"),
+
       NativeFunction("castBIGINT", {}, DataTypeVector{day_time_interval()}, int64(),
                      kResultNullIfNull, "castBIGINT_daytimeinterval"),
 
@@ -139,6 +142,12 @@ std::vector<NativeFunction> GetDateTimeFunctionRegistry() {
 
       NativeFunction("castDATE", {}, DataTypeVector{date64()}, date32(),
                      kResultNullIfNull, "castDATE_date64"),
+
+      NativeFunction("castTIMESTAMP", {}, DataTypeVector{date32()}, timestamp(),
+                     kResultNullIfNull, "castTIMESTAMP_date32"),
+
+      NativeFunction("castDATE", {}, DataTypeVector{timestamp()}, date32(),
+                     kResultNullIfNull, "castDATE32_timestamp"),
       DATE_TYPES(LAST_DAY_SAFE_NULL_IF_NULL, last_day, {});
 
   return date_time_fn_registry_;

cpp/src/gandiva/function_signature.cc

@@ -45,6 +45,14 @@ bool DataTypeEquals(const DataTypePtr& left, const DataTypePtr& right) {
         return (dleft != NULL) && (dright != NULL) &&
                (dleft->byte_width() == dright->byte_width());
       }
+      case arrow::Type::TIMESTAMP: {
+        // Signature for timestamp treated the same if both are with zone or without zone.
+        auto tleft = checked_cast<arrow::TimestampType *>(left.get());
+        auto tright = checked_cast<arrow::TimestampType *>(right.get());
+        return (tleft != NULL) && (tright != NULL) &&
+            (tleft->unit() == tright->unit()) &&
+            (tleft->timezone().empty() == tleft->timezone().empty());
+      }
       default:
         return left->Equals(right);
     }

cpp/src/gandiva/gdv_function_stubs.cc

@@ -311,7 +311,8 @@ CRC_FUNCTION(binary)
   INNER(date64)                             \
   INNER(date32)                             \
   INNER(time32)                             \
-  INNER(timestamp)
+  INNER(timestamp)                          \
+  INNER(timestampusutc)
 
 // Expand inner macro for all numeric types.
 #define SHA_VAR_LEN_PARAMS(INNER) \

cpp/src/gandiva/gdv_function_stubs.h

@@ -39,6 +39,7 @@ using gdv_date64 = int64_t;
 using gdv_date32 = int32_t;
 using gdv_time32 = int32_t;
 using gdv_timestamp = int64_t;
+using gdv_timestampusutc = int64_t;
 using gdv_utf8 = char*;
 using gdv_binary = char*;
 using gdv_day_time_interval = int64_t;

cpp/src/gandiva/jni/jni_common.cc

@@ -155,19 +155,26 @@ DataTypePtr ProtoTypeToTime64(const types::ExtGandivaType& ext_type) {
 }
 
 DataTypePtr ProtoTypeToTimestamp(const types::ExtGandivaType& ext_type) {
+  arrow::TimeUnit::type unit;
   switch (ext_type.timeunit()) {
     case types::SEC:
-      return arrow::timestamp(arrow::TimeUnit::SECOND);
+      unit = arrow::TimeUnit::SECOND;
+      break;
     case types::MILLISEC:
-      return arrow::timestamp(arrow::TimeUnit::MILLI);
+      unit = arrow::TimeUnit::MILLI;
+      break;
     case types::MICROSEC:
-      return arrow::timestamp(arrow::TimeUnit::MICRO);
+      unit = arrow::TimeUnit::MICRO;
+      break;
     case types::NANOSEC:
-      return arrow::timestamp(arrow::TimeUnit::NANO);
+      unit = arrow::TimeUnit::NANO;
+      break;
     default:
       std::cerr << "Unknown time unit: " << ext_type.timeunit() << " for timestamp\n";
       return nullptr;
   }
+  const std::string& zone_id = ext_type.timezone();
+  return arrow::timestamp(unit, zone_id);
 }
 
 DataTypePtr ProtoTypeToInterval(const types::ExtGandivaType& ext_type) {

cpp/src/gandiva/precompiled/arithmetic_ops.cc

@@ -41,6 +41,7 @@ extern "C" {
   INNER(NAME, date64, OP)           \
   INNER(NAME, date32, OP)           \
   INNER(NAME, timestamp, OP)        \
+  INNER(NAME, timestampusutc, OP)        \
   INNER(NAME, time32, OP)
 
 #define NUMERIC_DATE_TYPES(INNER, NAME, OP) \
@@ -362,6 +363,7 @@ NUMERIC_TYPES(IS_TRUE_OR_FALSE_NUMERIC, isfalse, !)
   INNER(date32)              \
   INNER(date64)              \
   INNER(timestamp)           \
+  INNER(timestampusutc)      \
   INNER(time32)
 
 #define NUMERIC_BOOL_DATE_FUNCTION(INNER) \

cpp/src/gandiva/precompiled/hash.cc

@@ -307,7 +307,8 @@ FORCE_INLINE gdv_int32 hash64_spark_int64_int32(gdv_int64 val, gdv_boolean is_va
   INNER(NAME, date64)                        \
   INNER(NAME, date32)                        \
   INNER(NAME, time32)                        \
-  INNER(NAME, timestamp)
+  INNER(NAME, timestamp)                     \
+  INNER(NAME, timestampusutc)
 
 NUMERIC_BOOL_DATE_TYPES(HASH32_OP, hash)
 NUMERIC_BOOL_DATE_TYPES(HASH32_OP, hash32)

cpp/src/gandiva/precompiled/time.cc

@@ -826,10 +826,18 @@ gdv_timestamp convertTimestampUnit_us(gdv_timestamp timestamp_in_micro) {
   return timestamp_in_micro / 1000;
 }
 
-gdv_date32 castDATE_date64(gdv_date64 date_in_millis) {
+gdv_date32 castDATE32_date64(gdv_date64 date_in_millis) {
   return static_cast<gdv_date32>(date_in_millis / (MILLIS_IN_DAY));
 }
 
+gdv_timestamp castTIMESTAMP_date32(gdv_date32 in_day) {
+  return static_cast<gdv_date32>(in_day * (MILLIS_IN_DAY));
+}
+
+gdv_date32 castDATE32_timestamp(gdv_timestamp timestamp_in_millis) {
+  return static_cast<gdv_date32>(timestamp_in_millis / (MILLIS_IN_DAY));
+}
+
 const char* castVARCHAR_timestamp_int64(gdv_int64 context, gdv_timestamp in,
                                         gdv_int64 length, gdv_int32* out_len) {
   gdv_int64 year = extractYear_timestamp(in);
@@ -896,6 +904,11 @@ gdv_int64 extractMillis_daytimeinterval(gdv_day_time_interval in) {
   return static_cast<gdv_int64>(millis);
 }
 
+FORCE_INLINE
+gdv_int64 castBIGINT_timestamp(gdv_timestamp in) {
+  return in;
+}
+
 FORCE_INLINE
 gdv_int64 castBIGINT_daytimeinterval(gdv_day_time_interval in) {
   return extractMillis_daytimeinterval(in) +