feat: Implement support for marshalling columns to velox Timestamp type.

velox/connectors/clp/ClpDataSource.cpp

@@ -93,6 +93,9 @@ void ClpDataSource::addFieldsRecursively(
       case TypeKind::ARRAY:
         clpColumnType = search_lib::ColumnType::Array;
         break;
+      case TypeKind::TIMESTAMP:
+        clpColumnType = search_lib::ColumnType::Timestamp;
+        break;
       default:
         VELOX_USER_FAIL("Type not supported: {}", columnType->name());
     }

velox/connectors/clp/search_lib/ClpCursor.cpp

@@ -213,6 +213,11 @@ ErrorCode ClpCursor::loadArchive() {
         case ColumnType::Array:
           columnDescriptor->set_matching_types(LiteralType::ArrayT);
           break;
+        case ColumnType::Timestamp:
+          columnDescriptor->set_matching_types(
+              LiteralType::EpochDateT | LiteralType::IntegerT |
+              LiteralType::FloatT);
+          break;
         default:
           break;
       }

velox/connectors/clp/search_lib/ClpCursor.h

@@ -49,7 +49,15 @@ enum class ErrorCode {
   Success
 };
 
-enum class ColumnType { String, Integer, Float, Array, Boolean, Unknown = -1 };
+enum class ColumnType {
+  Array,
+  Boolean,
+  Float,
+  Integer,
+  String,
+  Timestamp,
+  Unknown = -1
+};
 
 struct Field {
   ColumnType type;

velox/connectors/clp/search_lib/ClpVectorLoader.cpp

@@ -14,16 +14,113 @@
  * limitations under the License.
  */
 
+#include <cmath>
+#include <cstdint>
 #include <utility>
 
 #include "clp_s/ColumnReader.hpp"
+#include "clp_s/SchemaTree.hpp"
 
 #include "velox/connectors/clp/search_lib/ClpVectorLoader.h"
+#include "velox/type/Timestamp.h"
 #include "velox/vector/ComplexVector.h"
 #include "velox/vector/FlatVector.h"
 
 namespace facebook::velox::connector::clp::search_lib {
 
+namespace {
+
+enum class TimestampPrecision : uint8_t {
+  Seconds,
+  Milliseconds,
+  Microseconds,
+  Nanoseconds
+};
+
+/// Estimates the precision of an epoch timestamp as seconds, milliseconds,
+/// microseconds, or nanoseconds.
+///
+/// This heuristic relies on the fact that 1 year of epoch nanoseconds is
+/// approximately 1000 years of epoch microseconds and so on. This heuristic
+/// can be unreliable for timestamps sufficiently close to the epoch, but
+/// should otherwise be accurate for the next 1000 years.
+///
+/// Note: Future versions of the clp-s archive format will adopt a
+/// nanosecond-precision integer timestamp format (as opposed to the current
+/// format which allows other precisions), at which point we can remove this
+/// heuristic.
+///
+/// @param timestamp
+/// @return the estimated timestamp precision
+template <typename T>
+auto estimatePrecision(T timestamp) -> TimestampPrecision {
+  constexpr int64_t kEpochMilliseconds1971{31536000000};
+  constexpr int64_t kEpochMicroseconds1971{31536000000000};
+  constexpr int64_t kEpochNanoseconds1971{31536000000000000};
+  auto absTimestamp = timestamp >= 0 ? timestamp : -timestamp;
+
+  if (absTimestamp > kEpochNanoseconds1971) {
+    return TimestampPrecision::Nanoseconds;
+  } else if (absTimestamp > kEpochMicroseconds1971) {
+    return TimestampPrecision::Microseconds;
+  } else if (absTimestamp > kEpochMilliseconds1971) {
+    return TimestampPrecision::Milliseconds;
+  } else {
+    return TimestampPrecision::Seconds;
+  }
+}
+
+auto convertToVeloxTimestamp(double timestamp) -> Timestamp {
+  switch (estimatePrecision(timestamp)) {
+    case TimestampPrecision::Nanoseconds:
+      timestamp /= Timestamp::kNanosInSecond;
+      break;
+    case TimestampPrecision::Microseconds:
+      timestamp /= Timestamp::kMicrosecondsInSecond;
+      break;
+    case TimestampPrecision::Milliseconds:
+      timestamp /= Timestamp::kMillisecondsInSecond;
+      break;
+    case TimestampPrecision::Seconds:
+      break;
+  }
+  double seconds{std::floor(timestamp)};
+  double nanoseconds{(timestamp - seconds) * Timestamp::kNanosInSecond};
+  return Timestamp(
+      static_cast<int64_t>(seconds), static_cast<uint64_t>(nanoseconds));
+}
+
+auto convertToVeloxTimestamp(int64_t timestamp) -> Timestamp {
+  int64_t precisionDifference{Timestamp::kNanosInSecond};
+  switch (estimatePrecision(timestamp)) {
+    case TimestampPrecision::Nanoseconds:
+      break;
+    case TimestampPrecision::Microseconds:
+      precisionDifference =
+          Timestamp::kNanosInSecond / Timestamp::kNanosecondsInMicrosecond;
+      break;
+    case TimestampPrecision::Milliseconds:
+      precisionDifference =
+          Timestamp::kNanosInSecond / Timestamp::kNanosecondsInMillisecond;
+      break;
+    case TimestampPrecision::Seconds:
+      precisionDifference =
+          Timestamp::kNanosInSecond / Timestamp::kNanosInSecond;
+      break;
+  }
+  int64_t seconds{timestamp / precisionDifference};
+  int64_t nanoseconds{
+      (timestamp % precisionDifference) *
+      (Timestamp::kNanosInSecond / precisionDifference)};
+  if (nanoseconds < 0) {
+    seconds -= 1;
+    nanoseconds += Timestamp::kNanosInSecond;
+  }
+  return Timestamp(seconds, static_cast<uint64_t>(nanoseconds));
+}
+
+} // namespace
+
 ClpVectorLoader::ClpVectorLoader(
     clp_s::BaseColumnReader* columnReader,
     ColumnType nodeType,
@@ -57,6 +154,51 @@ void ClpVectorLoader::populateData(RowSet rows, VectorPtr vector) {
   }
 }
 
+template <clp_s::NodeType Type>
+void ClpVectorLoader::populateTimestampData(
+    RowSet rows,
+    FlatVector<facebook::velox::Timestamp>* vector) {
+  bool supportedNodeType{false};
+  switch (Type) {
+    case clp_s::NodeType::Float:
+    case clp_s::NodeType::Integer:
+    case clp_s::NodeType::DateString:
+      supportedNodeType = true;
+      break;
+    default:
+      break;
+  }
+  if (columnReader_ == nullptr || false == supportedNodeType) {
+    for (int vectorIndex : rows) {
+      vector->setNull(vectorIndex, true);
+    }
+    return;
+  }
+
+  for (int vectorIndex : rows) {
+    auto messageIndex = (*filteredRowIndices_)[vectorIndex];
+
+    if (clp_s::NodeType::Float == Type) {
+      auto reader = static_cast<clp_s::FloatColumnReader*>(columnReader_);
+      vector->set(
+          vectorIndex,
+          convertToVeloxTimestamp(
+              std::get<double>(reader->extract_value(messageIndex))));
+    } else if (clp_s::NodeType::Integer == Type) {
+      auto reader = static_cast<clp_s::Int64ColumnReader*>(columnReader_);
+      vector->set(
+          vectorIndex,
+          convertToVeloxTimestamp(
+              std::get<int64_t>(reader->extract_value(messageIndex))));
+    } else {
+      auto reader = static_cast<clp_s::DateStringColumnReader*>(columnReader_);
+      vector->set(
+          vectorIndex,
+          convertToVeloxTimestamp(reader->get_encoded_time(messageIndex)));
+    }
+  }
+}
+
 void ClpVectorLoader::loadInternal(
     RowSet rows,
     ValueHook* hook,
@@ -142,6 +284,23 @@ void ClpVectorLoader::loadInternal(
       }
       break;
     }
+    case ColumnType::Timestamp: {
+      auto timestampVector = vector->asFlatVector<Timestamp>();
+      if (nullptr != dynamic_cast<clp_s::Int64ColumnReader*>(columnReader_)) {
+        populateTimestampData<clp_s::NodeType::Integer>(rows, timestampVector);
+      } else if (
+          nullptr !=
+          dynamic_cast<clp_s::DateStringColumnReader*>(columnReader_)) {
+        populateTimestampData<clp_s::NodeType::DateString>(
+            rows, timestampVector);
+      } else if (
+          nullptr != dynamic_cast<clp_s::FloatColumnReader*>(columnReader_)) {
+        populateTimestampData<clp_s::NodeType::Float>(rows, timestampVector);
+      } else {
+        populateTimestampData<clp_s::NodeType::Unknown>(rows, timestampVector);
+      }
+      break;
+    }
     default:
       VELOX_FAIL("Unsupported column type");
   }
@@ -160,5 +319,18 @@ template void ClpVectorLoader::populateData<uint8_t>(
 template void ClpVectorLoader::populateData<std::string>(
     RowSet rows,
     FlatVector<StringView>* vector);
+template void ClpVectorLoader::populateTimestampData<clp_s::NodeType::Float>(
+    RowSet rows,
+    FlatVector<facebook::velox::Timestamp>* vector);
+template void ClpVectorLoader::populateTimestampData<clp_s::NodeType::Integer>(
+    RowSet rows,
+    FlatVector<facebook::velox::Timestamp>* vector);
+template void
+ClpVectorLoader::populateTimestampData<clp_s::NodeType::DateString>(
+    RowSet rows,
+    FlatVector<facebook::velox::Timestamp>* vector);
+template void ClpVectorLoader::populateTimestampData<clp_s::NodeType::Unknown>(
+    RowSet rows,
+    FlatVector<facebook::velox::Timestamp>* vector);
 
 } // namespace facebook::velox::connector::clp::search_lib

velox/connectors/clp/search_lib/ClpVectorLoader.h

@@ -16,7 +16,12 @@
 
 #pragma once
 
+#include "clp_s/ColumnReader.hpp"
+#include "clp_s/SchemaTree.hpp"
+
 #include "velox/connectors/clp/search_lib/ClpCursor.h"
+#include "velox/type/Timestamp.h"
+#include "velox/vector/FlatVector.h"
 #include "velox/vector/LazyVector.h"
 
 namespace clp_s {
@@ -45,6 +50,11 @@ class ClpVectorLoader : public VectorLoader {
   template <typename T, typename VectorPtr>
   void populateData(RowSet rows, VectorPtr vector);
 
+  template <clp_s::NodeType Type>
+  void populateTimestampData(
+      RowSet rows,
+      FlatVector<facebook::velox::Timestamp>* vector);
+
   clp_s::BaseColumnReader* columnReader_;
   ColumnType nodeType_;
   std::shared_ptr<std::vector<uint64_t>> filteredRowIndices_;