Add Spark CAST(double/float as timestamp) #12041
Conversation
✅ Deploy Preview for meta-velox canceled.
|
|
Hey @rui-mo @jinchengchenghh , can help review this? Thanks! |
There was a problem hiding this comment.
Perhaps also document this cast in https://github.com/facebookincubator/velox/blob/main/velox/docs/functions/spark/conversion.rst#cast-to-timestamp. Thanks.
| ToKind == TypeKind::TIMESTAMP) { | ||
| const auto castResult = | ||
| hooks_->castDoubleToTimestamp(static_cast<double>(inputRowValue)); | ||
| if (castResult.hasError()) { |
There was a problem hiding this comment.
Can we use 'setResultOrError'?
There was a problem hiding this comment.
We need to set Nulls if the value is Nan or Infinity, now I use the error type to distinguish this:
if (castResult.error().isUserError()) {
setError(castResult.error().message());
} else {
result->setNull(row, true);
}setResultOrError doesn't support handle error separately.
| Status::Invalid("Can not convert NaN or Infinity to timestamp")); | ||
| } | ||
| return Timestamp::fromMicrosNoError( | ||
| static_cast<int64_t>(value * Timestamp::kMicrosecondsInSecond)); |
There was a problem hiding this comment.
How is overflow handled? It looks different with the impl. of 'castIntToTimestamp'.
There was a problem hiding this comment.
Yes, different from 'castIntToTimestamp', Spark also doesn't handle overflow cases, can see
https://github.com/apache/spark/blob/fd86f85e181fc2dc0f50a096855acf83a6cc5d9c/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala#L675
There was a problem hiding this comment.
Do we need to handle it inside velox?
There was a problem hiding this comment.
Can you try a overflow case that value is max double？
Spark result is
spark-sql (default)> select cast(1.79769e+308 as timestamp);
+294247-01-10 04:00:54.775807
Not sure the result is same in Velox.
There was a problem hiding this comment.
Yes, you're correct, the overflow handling is not same with the Spark, fix them by add separate check, also add tests to cover it.
| @@ -291,6 +291,46 @@ TEST_F(SparkCastExprTest, intToTimestamp) { | |||
| testIntegralToTimestampCast<int32_t>(); | |||
| } | |||
|
|
|||
| TEST_F(SparkCastExprTest, doubleToTimestamp) { | |||
| testCast( | |||
| makeNullableFlatVector<double>({ | |||
There was a problem hiding this comment.
makeNullableFlatVector -> makeFlatVector
|
|
||
| TEST_F(SparkCastExprTest, floatToTimestamp) { | ||
| testCast( | ||
| makeNullableFlatVector<float>({ |
b7068cf to
d9069ab
Compare
| @@ -35,19 +35,37 @@ Expected<Timestamp> SparkCastHooks::castStringToTimestamp( | |||
| view.data(), view.size(), util::TimestampParseMode::kSparkCast); | |||
| } | |||
|
|
|||
| Expected<Timestamp> SparkCastHooks::castIntToTimestamp(int64_t seconds) const { | |||
| template <typename T> | |||
| Expected<Timestamp> SparkCastHooks::castNumberToTimestamp(T value) const { | |||
There was a problem hiding this comment.
Please still use the T seconds.
| @@ -75,5 +79,8 @@ class SparkCastHooks : public exec::CastHooks { | |||
| .skipTrailingZeros = true, | |||
| .zeroPaddingYear = true, | |||
| .dateTimeSeparator = ' '}; | |||
|
|
|||
| template <typename T> | |||
| Expected<Timestamp> castNumberToTimestamp(T value) const; | |||
There was a problem hiding this comment.
Please move the function before the class fields.
d9069ab to
10c055a
Compare
velox/expression/CastExpr-inl.h
Outdated
| const auto castResult = | ||
| hooks_->castDoubleToTimestamp(static_cast<double>(inputRowValue)); | ||
| if (castResult.hasError()) { | ||
| if (castResult.error().isUserError()) { |
There was a problem hiding this comment.
The error handling according to type seems to be Spark-specific. I'm wondering if it's possible to wrap it inside the cast hook.
There was a problem hiding this comment.
Since we have to return null for Nan and Inf, Looks Expected can't support return null value, so here I update to use Expected<std::optional<Timestamp>> to wrap the logic inside Spark implementation.
10c055a to
64a7230
Compare
Add Spark CAST (double/float as timestamp). The input value is treated as the
number of seconds since the epoch (1970-01-01 00:00:00 UTC).
Spark's implementation: https://github.com/apache/spark/blob/fd86f85e181fc2dc0f50a096855acf83a6cc5d9c/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala#L675