feat: Add support for Timestamp to Integral for Spark

velox/expression/CastExpr-inl.h

@@ -286,6 +286,14 @@ void CastExpr::applyCastKernel(
       return;
     }
 
+    if constexpr (
+        (FromKind == TypeKind::TIMESTAMP) && ToKind == TypeKind::BIGINT) {
+      const auto castResult =
+          hooks_->castTimestampToInt(inputRowValue);
+      setResultOrError(castResult, row);
+      return;
+    }
+
     // Optimize empty input strings casting by avoiding throwing exceptions.
     if constexpr (
         FromKind == TypeKind::VARCHAR || FromKind == TypeKind::VARBINARY) {

velox/expression/CastHooks.h

@@ -58,5 +58,8 @@ class CastHooks {
   virtual bool truncate() const = 0;
 
   virtual PolicyType getPolicy() const = 0;
+
+  // Cast timestamp to integer.
+  virtual Expected<int64_t> castTimestampToInt(Timestamp timestamp) const = 0;
 };
 } // namespace facebook::velox::exec

velox/expression/PrestoCastHooks.cpp

@@ -97,6 +97,11 @@ Expected<int32_t> PrestoCastHooks::castStringToDate(
   return util::fromDateString(dateString, util::ParseMode::kPrestoCast);
 }
 
+Expected<int64_t> PrestoCastHooks::castTimestampToInt(Timestamp timestamp) const {
+  return folly::makeUnexpected(
+      Status::UserError("Conversion to Timestamp is not supported"));
+}
+
 namespace {
 
 using double_conversion::StringToDoubleConverter;

velox/expression/PrestoCastHooks.h

@@ -32,6 +32,9 @@ class PrestoCastHooks : public CastHooks {
 
   Expected<Timestamp> castIntToTimestamp(int64_t seconds) const override;
 
+  // Cast timestamp to integer.
+  Expected<int64_t> castTimestampToInt(Timestamp timestamp) const override;
+
   // Uses standard cast mode to cast from string to date.
   Expected<int32_t> castStringToDate(
       const StringView& dateString) const override;

velox/functions/sparksql/specialforms/SparkCastHooks.cpp

@@ -50,6 +50,21 @@ Expected<Timestamp> SparkCastHooks::castIntToTimestamp(int64_t seconds) const {
   return Timestamp(seconds, 0);
 }
 
+Expected<int64_t> SparkCastHooks::castTimestampToInt(Timestamp timestamp) const {
+  static constexpr int64_t MICROS_PER_SECOND = 1'000'000;
+  int64_t micros = timestamp.toMicros();
+
+  // Replicate Spark conversion via floor division.
+  int64_t quotient = micros / MICROS_PER_SECOND;
+  int64_t remainder = micros % MICROS_PER_SECOND;
+
+  if (remainder != 0 && micros < 0) {
+    quotient -= 1;
+  }
+
+  return quotient;
+}
+
 Expected<int32_t> SparkCastHooks::castStringToDate(
     const StringView& dateString) const {
   // Allows all patterns supported by Spark:

velox/functions/sparksql/specialforms/SparkCastHooks.h

@@ -52,6 +52,9 @@ class SparkCastHooks : public exec::CastHooks {
   /// whitespaces before cast.
   StringView removeWhiteSpaces(const StringView& view) const override;
 
+  // Cast timestamp to integral.
+  Expected<int64_t> castTimestampToInt(Timestamp timestamp) const override;
+
   const TimestampToStringOptions& timestampToStringOptions() const override {
     return timestampToStringOptions_;
   }

velox/functions/sparksql/tests/SparkCastExprTest.cpp

@@ -750,5 +750,41 @@ TEST_F(SparkCastExprTest, bigintToBinary) {
        std::string("\x80\x00\x00\x00\x00\x00\x00\x00", 8)});
 }
 
+TEST_F(SparkCastExprTest, timestampToInt) {
+  std::vector<std::optional<Timestamp>> input = {
+      Timestamp(0, 0),
+      Timestamp(1, 0),
+      Timestamp(10, 0),
+      Timestamp(-1, 0),
+      Timestamp(-10, 0),
+      Timestamp(-1, 500000),
+      Timestamp(-2, 999999),
+      Timestamp(-10, 999999),
+      Timestamp(1, 999999),
+      Timestamp(-1, 1),
+      Timestamp(1234567, 500000),
+      Timestamp(-9876543, 1234),
+      std::nullopt,
+  };
+
+  std::vector<std::optional<int64_t>> expected = {
+      0,
+      1,
+      10,
+      -1,
+      -10,
+      -1,
+      -2,
+      -10,
+      1,
+      -1,
+      1234567,
+      -9876543,
+      std::nullopt,
+  };
+
+  testCast<Timestamp, int64_t>("bigint", input, expected);
+}
+
 } // namespace
 } // namespace facebook::velox::test