feat: Support int64_t value for Spark timestampadd function (facebookincubator#15459)

Summary:
Spark 4.0 change `value` from int32_t to int64_t, this PR is quickly to support it.
https://github.com/apache/spark/blob/29434ea766b0fc3c3bf6eaadb43a8f931133649e/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/datetimeExpressions.scala#L3442

Pull Request resolved: facebookincubator#15459

Reviewed By: peterenescu

Differential Revision: D87562354

Pulled By: kagamiori

fbshipit-source-id: d748bff943af6b196225c33528b24f8c5a336fdc

Author: zml1206

velox/docs/functions/spark/datetime.rst

@@ -287,7 +287,7 @@ These functions support TIMESTAMP and DATE input types.
 
 .. spark:function:: timestampadd(unit, value, timestamp) -> timestamp
 
-    Adds an interval ``value`` of type ``unit`` to ``timestamp``.
+    Adds an int or bigint interval ``value`` of type ``unit`` to ``timestamp``.
     Subtraction can be performed by using a negative ``value``.
     Throws exception if ``unit`` is invalid.
     ``unit`` is case insensitive and must be one of the following:

velox/functions/lib/DateTimeUtil.h

@@ -292,11 +292,12 @@ addToTimestamp(const Timestamp& timestamp, DateTimeUnit unit, int32_t value) {
       outTimestamp;
 
   switch (unit) {
-    // Year, quarter or month are not uniformly incremented in terms of number
-    // of days. So we treat them differently.
+    // Year, quarter, month or week are not uniformly incremented in terms of
+    // number of days. So we treat them differently.
     case DateTimeUnit::kYear:
     case DateTimeUnit::kQuarter:
     case DateTimeUnit::kMonth:
+    case DateTimeUnit::kWeek:
     case DateTimeUnit::kDay: {
       const int32_t inDate =
           std::chrono::duration_cast<date::days>(inTimestamp.time_since_epoch())
@@ -322,15 +323,6 @@ addToTimestamp(const Timestamp& timestamp, DateTimeUnit unit, int32_t value) {
       outTimestamp = inTimestamp + std::chrono::milliseconds(value);
       break;
     }
-    case DateTimeUnit::kWeek: {
-      const int32_t inDate =
-          std::chrono::duration_cast<date::days>(inTimestamp.time_since_epoch())
-              .count();
-      const int32_t outDate = addToDate(inDate, DateTimeUnit::kDay, 7 * value);
-
-      outTimestamp = inTimestamp + date::days(outDate - inDate);
-      break;
-    }
     default:
       VELOX_UNREACHABLE("Unsupported datetime unit");
   }

velox/functions/sparksql/DateTimeFunctions.h

@@ -1053,11 +1053,12 @@ template <typename T>
 struct TimestampAddFunction {
   VELOX_DEFINE_FUNCTION_TYPES(T);
 
+  template <typename TInput>
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
       const arg_type<Varchar>* unitString,
-      const int32_t* /*value*/,
+      const TInput* /*value*/,
       const arg_type<Timestamp>* /*timestamp*/) {
     VELOX_USER_CHECK_NOT_NULL(unitString);
     std::string unitStr(*unitString);
@@ -1071,10 +1072,11 @@ struct TimestampAddFunction {
     sessionTimeZone_ = getTimeZoneFromConfig(config);
   }
 
+  template <typename TInput>
   FOLLY_ALWAYS_INLINE void call(
       out_type<Timestamp>& result,
       const arg_type<Varchar>& /*unitString*/,
-      const int32_t value,
+      const TInput value,
       const arg_type<Timestamp>& timestamp) {
     const auto unit = unit_.value();
     result = addToTimestamp(unit, value, timestamp, sessionTimeZone_);

velox/functions/sparksql/registration/RegisterDatetime.cpp

@@ -110,6 +110,12 @@ void registerDatetimeFunctions(const std::string& prefix) {
       Varchar,
       int32_t,
       Timestamp>({prefix + "timestampadd"});
+  registerFunction<
+      TimestampAddFunction,
+      Timestamp,
+      Varchar,
+      int64_t,
+      Timestamp>({prefix + "timestampadd"});
   registerFunction<MonthsBetweenFunction, double, Timestamp, Timestamp, bool>(
       {prefix + "months_between"});
 }

velox/functions/sparksql/tests/DateTimeFunctionsTest.cpp

@@ -1705,21 +1705,39 @@ TEST_F(DateTimeFunctionsTest, timestampadd) {
         fmt::format("timestampadd('{}', c0, c1)", unit), value, timestamp);
   };
 
+  const auto timestampaddBigint = [&](const std::string& unit,
+                                      std::optional<int64_t> value,
+                                      std::optional<Timestamp> timestamp) {
+    return evaluateOnce<Timestamp>(
+        fmt::format("timestampadd('{}', c0, c1)", unit), value, timestamp);
+  };
+
   // Check null behaviors.
   EXPECT_EQ(std::nullopt, timestampadd("second", 1, std::nullopt));
   EXPECT_EQ(std::nullopt, timestampadd("month", std::nullopt, Timestamp(0, 0)));
+  EXPECT_EQ(
+      std::nullopt, timestampaddBigint("month", std::nullopt, Timestamp(0, 0)));
 
   // Check invalid units.
   VELOX_ASSERT_THROW(
       timestampadd("invalid_unit", 1, Timestamp(0, 0)),
       "Unsupported datetime unit: invalid_unit");
+  VELOX_ASSERT_THROW(
+      timestampaddBigint("invalid_unit", 1, Timestamp(0, 0)),
+      "Unsupported datetime unit: invalid_unit");
 
   EXPECT_EQ(
       Timestamp(1551348061, 999) /*2019-02-28 10:01:01.000000*/,
       timestampadd(
           "microsecond",
           60 * 1000000 + 500,
           Timestamp(1551348000, 999'500'999) /*2019-02-28 10:00:00.999500*/));
+  EXPECT_EQ(
+      Timestamp(1551348061, 999) /*2019-02-28 10:01:01.000000*/,
+      timestampaddBigint(
+          "microsecond",
+          60 * 1000000 + 500,
+          Timestamp(1551348000, 999'500'999) /*2019-02-28 10:00:00.999500*/));
   EXPECT_EQ(
       Timestamp(1551348061, 999'999) /*2019-02-28 10:01:01.000*/,
       timestampadd(