feat(dwio): Add DATE to TIMESTAMP widening coercion in DWIO readers

Reading a DATE column as TIMESTAMP previously failed at
schema-compatibility validation because DateType inherits from
IntegerType (kind() == INTEGER), and integer→timestamp is not a numeric
widening. DWIO readers now accept the coercion and emit Timestamp
(days × 86400 seconds, 0 nanos) at read time.

The widening is wired in three places. TypeUtils.cpp short-circuits
isCompatible() when the file type is DATE and the requested kind is
TIMESTAMP. SelectiveColumnReader gains convertDateToTimestampValues(),
dispatched from getIntValues() when the requested kind is TIMESTAMP and
the file type is DATE — it follows the same sourceRows iteration pattern
as upcastScalarValues but applies the days × 86400 conversion and stores
Timestamp structs. ParquetReader.cpp's convertType() for
ConvertedType::DATE now also accepts TIMESTAMP as a valid requested type,
in addition to DATE itself.

Test plan:
- velox/dwio/parquet/tests/reader/ParquetReaderTest.cpp:
  - readDateColumnAsTimestamp
  - readDateColumnAsTimestampWithNulls
  - readRealColumnAsDouble, readRealColumnAsDoubleWithNulls — coverage for pre-existing FLOAT→DOUBLE widening, no production change in this commit.
- velox/dwio/dwrf/test/ReaderTest.cpp:
  - readDateColumnAsTimestamp
  - readDateColumnAsTimestampWithNulls

Author: yingsu00

velox/dwio/common/SelectiveColumnReader.cpp

@@ -303,6 +303,25 @@ void SelectiveColumnReader::getIntValues(
           VELOX_FAIL("Unsupported value size: {}", valueSize_);
       }
       break;
+    case TypeKind::TIMESTAMP:
+      // DateType inherits from IntegerType (kind() == INTEGER), and DWRF
+      // erases DATE to plain INT during write -- so the genuine-DATE case
+      // and the post-DWRF-roundtrip case both present as kind() == INTEGER
+      // here.  Conversion (days * 86400) is the same in both.
+      VELOX_CHECK_EQ(
+          fileType_->type()->kind(),
+          TypeKind::INTEGER,
+          "TIMESTAMP output requires an INTEGER-kind file type (DATE or INTEGER), got: {}",
+          fileType_->type()->toString());
+      if (valueSize_ == sizeof(Timestamp)) {
+        // A prior convertDateToTimestampValues call already bulk-transmuted
+        // the int32 days buffer to Timestamps; further extractions are plain
+        // Timestamp compactions.
+        getFlatValues<Timestamp, Timestamp>(rows, result, TIMESTAMP());
+      } else {
+        convertDateToTimestampValues(rows, result);
+      }
+      break;
     default:
       VELOX_FAIL(
           "Not a valid type for integer reader: {}", requestedType->toString());
@@ -452,6 +471,108 @@ void SelectiveColumnReader::compactScalarValues<bool, bool>(
   values_->setSize(bits::nbytes(numValues_));
 }
 
+void SelectiveColumnReader::convertDateToTimestampValues(
+    const RowSet& rows,
+    VectorPtr* result,
+    bool isFinal) {
+  VELOX_CHECK_EQ(valueSize_, sizeof(int32_t));
+  VELOX_CHECK(mayGetValues_);
+
+  if (allNull_) {
+    *result = std::make_shared<ConstantVector<Timestamp>>(
+        pool_, rows.size(), true, TIMESTAMP(), Timestamp());
+    if (isFinal) {
+      mayGetValues_ = false;
+    }
+    return;
+  }
+  VELOX_CHECK_NOT_NULL(values_);
+
+  static constexpr int64_t kSecondsPerDay{86'400};
+
+  // Bulk path: this single call asks for at least half of the read values.
+  // Transmute all numValues_ days to Timestamps in place (amortizing the
+  // conversion), transition to State B, then emit the requested slice via
+  // standard Timestamp extraction. Iterates backwards so the wider
+  // Timestamp writes don't clobber int32 source bytes we haven't read yet.
+  if (rows.size() >= numValues_ / 2) {
+    ensureValuesCapacity<Timestamp>(numValues_, true);
+    const auto* rawDays = reinterpret_cast<const int32_t*>(rawValues_);
+    auto* timestamps = reinterpret_cast<Timestamp*>(rawValues_);
+    for (auto i = numValues_; i-- > 0;) {
+      timestamps[i] =
+          Timestamp(kSecondsPerDay * static_cast<int64_t>(rawDays[i]), 0);
+    }
+    valueSize_ = sizeof(Timestamp);
+    values_->setSize(numValues_ * sizeof(Timestamp));
+    getFlatValues<Timestamp, Timestamp>(rows, result, TIMESTAMP(), isFinal);
+    return;
+  }
+
+  // Small-subset path: leave the int32 days buffer intact and emit a
+  // freshly-allocated Timestamp vector for just rows.size(). Multi-call
+  // safe because values_, valueSize_, valueRows_ are untouched.
+  const auto sourceRows = valueRows_.empty()
+      ? (outputRows_.empty() ? RowSet(inputRows_) : RowSet(outputRows_))
+      : RowSet(valueRows_);
+  const auto* rawDays = reinterpret_cast<const int32_t*>(rawValues_);
+
+  auto timestampValues = AlignedBuffer::allocate<Timestamp>(
+      rows.size() + simd::kPadding / sizeof(Timestamp), pool_);
+  auto* timestamps = timestampValues->asMutable<Timestamp>();
+
+  const auto* moveNullsFrom = shouldMoveNulls(rows);
+  BufferPtr localNullsBuffer;
+  uint64_t* localNulls = nullptr;
+  if (moveNullsFrom) {
+    localNullsBuffer = AlignedBuffer::allocate<bool>(rows.size(), pool_);
+    localNulls = localNullsBuffer->asMutable<uint64_t>();
+  }
+
+  size_t sourceIndex{0};
+  if (moveNullsFrom) {
+    for (vector_size_t rowIndex = 0; rowIndex < rows.size();) {
+      const auto begin = rowIndex;
+      const auto end = std::min<vector_size_t>(begin + 64, rows.size());
+      uint64_t nulls = 0;
+      for (; rowIndex < end; ++rowIndex) {
+        while (sourceRows[sourceIndex] < rows[rowIndex]) {
+          ++sourceIndex;
+        }
+        VELOX_DCHECK_EQ(sourceRows[sourceIndex], rows[rowIndex]);
+        timestamps[rowIndex] = Timestamp(
+            kSecondsPerDay * static_cast<int64_t>(rawDays[sourceIndex]), 0);
+        nulls |=
+            static_cast<uint64_t>(bits::isBitSet(moveNullsFrom, sourceIndex))
+            << (rowIndex - begin);
+        ++sourceIndex;
+      }
+      localNulls[begin / 64] = nulls;
+    }
+  } else {
+    for (vector_size_t rowIndex = 0; rowIndex < rows.size(); ++rowIndex) {
+      while (sourceRows[sourceIndex] < rows[rowIndex]) {
+        ++sourceIndex;
+      }
+      VELOX_DCHECK_EQ(sourceRows[sourceIndex], rows[rowIndex]);
+      timestamps[rowIndex] = Timestamp(
+          kSecondsPerDay * static_cast<int64_t>(rawDays[sourceIndex++]), 0);
+    }
+  }
+
+  *result = std::make_shared<FlatVector<Timestamp>>(
+      pool_,
+      TIMESTAMP(),
+      moveNullsFrom ? localNullsBuffer : resultNulls(),
+      rows.size(),
+      std::move(timestampValues),
+      std::vector<BufferPtr>{});
+
+  if (isFinal) {
+    mayGetValues_ = false;
+  }
+}
+
 char* SelectiveColumnReader::copyStringValue(std::string_view value) {
   uint64_t size = value.size();
   if (stringBuffers_.empty() || rawStringUsed_ + size > rawStringSize_) {

velox/dwio/common/SelectiveColumnReader.h

@@ -595,6 +595,23 @@ class SelectiveColumnReader {
   template <typename T, typename TVector>
   void upcastScalarValues(const RowSet& rows);
 
+  /// Reads DATE (int32 days-since-epoch) values and converts them to
+  /// Timestamp (seconds = days × 86400, nanoseconds = 0).
+  /// Emits Timestamps for 'rows' from a buffer of int32 days-since-epoch read
+  /// by prepareRead<int32_t>. Reentrant: can be called multiple times for
+  /// disjoint subsets of 'rows' from the same read, mirroring getFlatValues.
+  /// If rows.size() >= numValues_/2 the int32 buffer is transmuted to
+  /// Timestamps in place (amortizing the conversion) and valueSize_ becomes
+  /// sizeof(Timestamp); subsequent extractions should use
+  /// getFlatValues<Timestamp, Timestamp> directly. Otherwise emits a fresh
+  /// Timestamp vector and leaves the source buffer intact. Caller (getIntValues)
+  /// dispatches on valueSize_ before calling. Pass isFinal=true on the last
+  /// call to release the mayGetValues_ lock.
+  void convertDateToTimestampValues(
+      const RowSet& rows,
+      VectorPtr* result,
+      bool isFinal = false);
+
   // For complex type column, we need to compact only nulls if the rows are
   // shrinked.  Child fields are handled recursively in their own column
   // readers.

velox/dwio/common/TypeUtils.cpp

@@ -130,7 +130,18 @@ void checkTypeCompatibility(
     const FKind& kind,
     const FShouldRead& shouldRead,
     const std::function<std::string()>& exceptionMessageCreator) {
-  if (shouldRead(to) && !isCompatible(from.kind(), kind(to))) {
+  // Special case: DATE (stored as int32 days-since-epoch) may be widened to
+  // TIMESTAMP (seconds-since-epoch).  The semantic conversion (days × 86400)
+  // is performed at read time by convertDateToTimestampValues().  DWRF erases
+  // DATE to plain INT in the proto schema during write, so on the read side
+  // the file-side kind() is INTEGER in both the genuine-DATE case (DateType
+  // inherits from IntegerType, kind() == INTEGER) and the post-roundtrip
+  // case.  Match on kind() == INTEGER so both shapes resolve.
+  const bool isIntToTimestamp = from.kind() == TypeKind::INTEGER &&
+      kind(to) == TypeKind::TIMESTAMP;
+
+  if (shouldRead(to) && !isIntToTimestamp &&
+      !isCompatible(from.kind(), kind(to))) {
     VELOX_SCHEMA_MISMATCH_ERROR(
         fmt::format(
             "{}, From Kind: {}, To Kind: {}",