fix int96

Author: rui-mo

velox/dwio/common/SelectiveColumnReader.cpp

@@ -190,6 +190,10 @@ void SelectiveColumnReader::getIntValues(
         getFlatValues<UnscaledLongDecimal, UnscaledLongDecimal>(
             rows, result, requestedType);
         break;
+      case TypeKind::TIMESTAMP:
+        getFlatValues<Timestamp, Timestamp>(
+            rows, result, requestedType);
+        break;
       case TypeKind::BIGINT:
         switch (valueSize_) {
           case 8:

velox/dwio/parquet/reader/PageReader.cpp

@@ -414,8 +414,9 @@ void PageReader::prepareDictionary(const PageHeader& pageHeader) {
       break;
     }
     case thrift::Type::INT96: {
+      auto numVeloxBytes = dictionary_.numValues * sizeof(Timestamp);
+      dictionary_.values = AlignedBuffer::allocate<char>(numVeloxBytes, &pool_);
       auto numBytes = dictionary_.numValues * sizeof(int96_t);
-      dictionary_.values = AlignedBuffer::allocate<char>(numBytes, &pool_);
       if (pageData_) {
         memcpy(dictionary_.values->asMutable<char>(), pageData_, numBytes);
       } else {
@@ -426,6 +427,23 @@ void PageReader::prepareDictionary(const PageHeader& pageHeader) {
             bufferStart_,
             bufferEnd_);
       }
+      // Expand the Parquet type length values to Velox type length.
+      // We start from the end to allow in-place expansion.
+      auto values = dictionary_.values->asMutable<Timestamp>();
+      auto parquetValues = dictionary_.values->asMutable<char>();
+      constexpr int64_t JULIAN_TO_UNIX_EPOCH_DAYS = 2440588LL;
+      constexpr int64_t SECONDS_PER_DAY = 86400LL;
+      for (auto i = dictionary_.numValues - 1; i >= 0; --i) {
+        // Convert the timestamp into seconds and nanos since the Unix epoch,
+        // 00:00:00.000000 on 1 January 1970.
+        uint64_t nanos;
+        memcpy(&nanos, parquetValues + i * sizeof(int96_t), sizeof(uint64_t));
+        int32_t days;
+        memcpy(&days, parquetValues + i * sizeof(int96_t) + + sizeof(uint64_t),
+            sizeof(int32_t));
+        values[i] = Timestamp(
+            (days - JULIAN_TO_UNIX_EPOCH_DAYS) * SECONDS_PER_DAY, nanos);
+      }
       break;
     }
     case thrift::Type::BYTE_ARRAY: {

velox/dwio/parquet/reader/TimestampColumnReader.h

@@ -41,82 +41,6 @@ class TimestampColumnReader : public IntegerColumnReader {
     prepareRead<int128_t>(offset, rows, nullptr);
     readCommon<IntegerColumnReader>(rows);
   }
-
-  void getValues(RowSet rows, VectorPtr* result) override {
-    auto type = nodeType_->type;
-    VELOX_CHECK(type->kind() == TypeKind::TIMESTAMP, "Timestamp expected.");
-    VELOX_CHECK_NE(valueSize_, kNoValueSize);
-    VELOX_CHECK(mayGetValues_);
-    if (allNull_) {
-      *result = std::make_shared<ConstantVector<Timestamp>>(
-          &memoryPool_,
-          rows.size(),
-          true,
-          type,
-          Timestamp(),
-          SimpleVectorStats<Timestamp>{},
-          sizeof(Timestamp) * rows.size());
-      return;
-    }
-    VELOX_CHECK_LE(rows.size(), numValues_);
-    VELOX_CHECK(!rows.empty());
-    if (!values_) {
-      return;
-    }
-
-    auto tsValues = AlignedBuffer::allocate<Timestamp>(
-        rows.size(), &memoryPool_, Timestamp());
-    auto* valuesPtr = tsValues->asMutable<Timestamp>();
-    char* rawValues = reinterpret_cast<char*>(rawValues_);
-
-    vector_size_t rowIndex = 0;
-    auto nextRow = rows[rowIndex];
-    bool moveNulls = shouldMoveNulls(rows);
-    bool emptyOutputRows = outputRows_.size() == 0;
-    for (size_t i = 0; i < numValues_; i++) {
-      if (!emptyOutputRows && outputRows_[i] < nextRow) {
-        continue;
-      }
-      VELOX_DCHECK(emptyOutputRows || (outputRows_[i] == nextRow));
-
-      // Convert the timestamp into seconds and nanos since the Unix epoch,
-      // 00:00:00.000000 on 1 January 1970.
-      uint64_t nanos;
-      memcpy(&nanos, rawValues + nextRow * sizeof(int96_t), sizeof(uint64_t));
-      int32_t days;
-      memcpy(
-          &days,
-          rawValues + nextRow * sizeof(int96_t) + sizeof(uint64_t),
-          sizeof(int32_t));
-      valuesPtr[rowIndex] = Timestamp(
-          (days - JULIAN_TO_UNIX_EPOCH_DAYS) * SECONDS_PER_DAY, nanos);
-
-      if (moveNulls && rowIndex != i) {
-        bits::setBit(
-            rawResultNulls_, rowIndex, bits::isBitSet(rawResultNulls_, i));
-      }
-      if (!emptyOutputRows) {
-        outputRows_[rowIndex] = nextRow;
-      }
-      rowIndex++;
-      if (rowIndex >= rows.size()) {
-        break;
-      }
-      nextRow = rows[rowIndex];
-    }
-
-    BufferPtr nulls = anyNulls_
-        ? (returnReaderNulls_ ? nullsInReadRange_ : resultNulls_)
-        : nullptr;
-
-    *result = std::make_shared<FlatVector<Timestamp>>(
-        &memoryPool_,
-        type,
-        nulls,
-        rows.size(),
-        tsValues,
-        std::move(stringBuffers_));
-  }
 };
 
 } // namespace facebook::velox::parquet

velox/type/Type.h

@@ -46,8 +46,9 @@
 namespace facebook::velox {
 
 using int128_t = __int128_t;
-struct int96_t {
-  int32_t val[3];
+struct __attribute__((__packed__)) int96_t {
+  int32_t days;
+  uint64_t nanos;
 };
 
 /// Velox type system supports a small set of SQL-compatible composeable types: