serialization of timestamp in spilling (facebookincubator#6838)

Summary:
Pull Request resolved: facebookincubator#6838

Fix serialization of timestamp in spilling. TimeStamp does not always fit in
one int64_t as nanos. More information about the issue here.
facebookincubator#6837.

This fixes several fuzzer issues such as
facebookincubator#6833

Reviewed By: mbasmanova

Differential Revision: D49796384

fbshipit-source-id: c1c45ebc0e3d159e1fc2db1cd315dae5d1957d31

velox/exec/tests/SpillTest.cpp

@@ -13,16 +13,19 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-#include "velox/exec/Spill.h"
+
 #include <gtest/gtest.h>
 #include <algorithm>
 #include <memory>
+
 #include "velox/common/base/RuntimeMetrics.h"
 #include "velox/common/base/tests/GTestUtils.h"
 #include "velox/common/file/FileSystems.h"
 #include "velox/exec/OperatorUtils.h"
+#include "velox/exec/Spill.h"
 #include "velox/exec/tests/utils/TempDirectoryPath.h"
 #include "velox/serializers/PrestoSerializer.h"
+#include "velox/type/Timestamp.h"
 #include "velox/vector/tests/utils/VectorTestBase.h"
 
 using namespace facebook::velox;
@@ -424,7 +427,9 @@ TEST_P(SpillTest, spillTimestamp) {
       Timestamp{12, 0},
       Timestamp{0, 17'123'456},
       Timestamp{1, 17'123'456},
-      Timestamp{-1, 17'123'456}};
+      Timestamp{-1, 17'123'456},
+      Timestamp{Timestamp::kMaxSeconds, Timestamp::kMaxNanos},
+      Timestamp{Timestamp::kMinSeconds, 0}};
 
   SpillState state(
       spillPath,

velox/serializers/PrestoSerializer.cpp

@@ -226,8 +226,9 @@ void readValues<Timestamp>(
 }
 
 Timestamp readLosslessTimestamp(ByteStream* source) {
-  int64_t nanos = source->read<int64_t>();
-  return Timestamp::fromNanos(nanos);
+  int64_t seconds = source->read<int64_t>();
+  uint64_t nanos = source->read<uint64_t>();
+  return Timestamp(seconds, nanos);
 }
 
 void readLosslessTimestampValues(
@@ -1014,7 +1015,8 @@ template <>
 void VectorStream::append(folly::Range<const Timestamp*> values) {
   if (useLosslessTimestamp_) {
     for (auto& value : values) {
-      appendOne(value.toNanos());
+      appendOne(value.getSeconds());
+      appendOne(value.getNanos());
     }
   } else {
     for (auto& value : values) {