feat: Add timestamp type support for kv-ir splits.

velox/connectors/clp/search_lib/BaseClpCursor.cpp

@@ -30,6 +30,73 @@ using namespace clp_s::search::ast;
 
 namespace facebook::velox::connector::clp::search_lib {
 
+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));
+}
+
 void BaseClpCursor::executeQuery(
     const std::string& query,
     const std::vector<Field>& outputColumns) {

velox/connectors/clp/search_lib/BaseClpCursor.h

@@ -60,6 +60,43 @@ struct Field {
   std::string name;
 };
 
+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;
+
+/// Converts a float value into a Velox timestamp.
+///
+/// @param timestamp the input timestamp as a float
+/// @return the corresponding Velox timestamp
+auto convertToVeloxTimestamp(double timestamp) -> Timestamp;
+
+/// Converts an integer value into a Velox timestamp.
+///
+/// @param timestamp the input timestamp as an integer
+/// @return the corresponding Velox timestamp
+auto convertToVeloxTimestamp(int64_t timestamp) -> Timestamp;
+
 /// A query execution interface that manages the lifecycle of a query on a CLP-S
 /// split (archive or IR), including parsing and validating the query, loading
 /// the relevant splits, applying filters, and iterating over the results. It

velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp

@@ -28,99 +28,6 @@
 
 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
-
 ClpArchiveVectorLoader::ClpArchiveVectorLoader(
     clp_s::BaseColumnReader* columnReader,
     ColumnType nodeType,

velox/connectors/clp/search_lib/ir/CMakeLists.txt

@@ -22,12 +22,13 @@ velox_add_library(
 
 velox_link_libraries(
   clp-s-ir-search
-  PUBLIC clp_s::archive_reader
+  PUBLIC clp_s::archive_reader clp_s::archive_writer
   PRIVATE
     # Once the IR-stream-related targets are exported, this should be updated to
     # use more fine-grained dependencies.
     clp_s::clp_dependencies
     clp_s::io
     clp_s::search
+    clp_s::search::ast
     clp_s::search::kql
     velox_vector)

velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp

@@ -59,10 +59,10 @@ VectorPtr ClpIrCursor::createVector(
     const TypePtr& vectorType,
     size_t vectorSize) {
   VELOX_CHECK_EQ(
-      projectedColumnIdxNodeIdMap_.size(),
+      projectedColumnIdxNodeIdsMap_.size(),
       outputColumns_.size(),
       "Projected columns size {} does not match fields size {}",
-      projectedColumnIdxNodeIdMap_.size(),
+      projectedColumnIdxNodeIdsMap_.size(),
       outputColumns_.size());
   return createVectorHelper(pool, vectorType, vectorSize);
 }
@@ -135,7 +135,8 @@ ClpIrCursor::splitFieldsToNamesAndTypes() const {
       case ColumnType::Timestamp:
         // TODO: IR timestamp support pending; constrain to Unknown to avoid
         // mismatched projections.
-        literalType = search::ast::LiteralType::EpochDateT;
+        literalType = search::ast::LiteralType::FloatT |
+            search::ast::LiteralType::IntegerT;
         break;
       default:
         literalType = search::ast::LiteralType::UnknownT;
@@ -189,22 +190,23 @@ VectorPtr ClpIrCursor::createVectorHelper(
       readerIndex_, outputColumns_.size(), "Reader index out of bounds");
   auto projectedColumn = outputColumns_[readerIndex_];
   auto projectedColumnType = projectedColumn.type;
-  auto it = projectedColumnIdxNodeIdMap_.find(readerIndex_);
-  bool isResolved = it != projectedColumnIdxNodeIdMap_.end();
-  ::clp::ffi::SchemaTree::Node::id_t projectedColumnNodeId;
+  auto it = projectedColumnIdxNodeIdsMap_.find(readerIndex_);
+  bool isResolved = it != projectedColumnIdxNodeIdsMap_.end();
+  std::vector<::clp::ffi::SchemaTree::Node::id_t> projectedColumnNodeIds;
   if (isResolved) {
-    projectedColumnNodeId = it->second;
+    projectedColumnNodeIds = it->second;
   }
   readerIndex_++;
   return std::make_shared<LazyVector>(
       pool,
       vectorType,
       vectorSize,
       std::make_unique<ClpIrVectorLoader>(
+          irDeserializer_->get_ir_unit_handler().getFilteredLogEvents(),
           isResolved,
-          projectedColumnType,
-          projectedColumnNodeId,
-          irDeserializer_->get_ir_unit_handler().getFilteredLogEvents()),
+          projectedColumnNodeIds,
+          projectedColumn.name,
+          projectedColumnType),
       std::move(vector));
 }
 

velox/connectors/clp/search_lib/ir/ClpIrCursor.h

@@ -59,8 +59,15 @@ class ClpIrCursor final : public BaseClpCursor {
               [[maybe_unused]] std::pair<std::string_view, size_t>
                   projected_key_and_index)
       -> ystdlib::error_handling::Result<void> {
-    projectedColumnIdxNodeIdMap_.insert(
-        {projected_key_and_index.second, nodeId});
+    auto it =
+        projectedColumnIdxNodeIdsMap_.find(projected_key_and_index.second);
+    if (it == projectedColumnIdxNodeIdsMap_.end()) {
+      projectedColumnIdxNodeIdsMap_.insert(
+          {projected_key_and_index.second,
+           std::vector<::clp::ffi::SchemaTree::Node::id_t>{nodeId}});
+      return ystdlib::error_handling::success();
+    }
+    it->second.emplace_back(nodeId);
     return ystdlib::error_handling::success();
   };
   using QueryHandlerType = ::clp::ffi::ir_stream::search::QueryHandler<
@@ -70,8 +77,8 @@ class ClpIrCursor final : public BaseClpCursor {
       ::clp::ffi::ir_stream::Deserializer<ClpIrUnitHandler, QueryHandlerType>>
       irDeserializer_;
   std::shared_ptr<::clp::ReaderInterface> irReader_{nullptr};
-  std::unordered_map<size_t, ::clp::ffi::SchemaTree::Node::id_t>
-      projectedColumnIdxNodeIdMap_;
+  std::unordered_map<size_t, std::vector<::clp::ffi::SchemaTree::Node::id_t>>
+      projectedColumnIdxNodeIdsMap_;
   size_t readerIndex_{0};
 
   std::vector<

velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cpp

@@ -14,8 +14,10 @@
  * limitations under the License.
  */
 
-#include "velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h"
+#include <chrono>
+
 #include "velox/connectors/clp/search_lib/BaseClpCursor.h"
+#include "velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h"
 
 namespace facebook::velox::connector::clp::search_lib {
 
@@ -33,12 +35,21 @@ void ClpIrVectorLoader::loadInternal(
     auto& logEvent = filteredLogEvents_->at(vectorIndex);
     // TODO: also need to support auto-generated keys
     auto userGenNodeIdValueMap = logEvent->get_user_gen_node_id_value_pairs();
-    auto const value_it{userGenNodeIdValueMap.find(nodeId_)};
-    if (userGenNodeIdValueMap.end() == value_it ||
-        false == value_it->second.has_value()) {
+    std::unordered_map<id_t, std::optional<::clp::ffi::Value>>::iterator
+        valueIt;
+    ::clp::ffi::SchemaTree::Node::id_t nodeId;
+    for (size_t i{0}; i < nodeIds_.size(); ++i) {
+      valueIt = userGenNodeIdValueMap.find(nodeIds_[i]);
+      if (valueIt != userGenNodeIdValueMap.end()) {
+        nodeId = nodeIds_[i];
+        break;
+      }
+    }
+    if (userGenNodeIdValueMap.end() == valueIt ||
+        false == valueIt->second.has_value()) {
       continue;
     }
-    auto const& value{value_it->second};
+    auto const& value{valueIt->second};
     switch (nodeType_) {
       case ColumnType::String: {
         auto stringVector = vector->asFlatVector<StringView>();
@@ -89,6 +100,27 @@ void ClpIrVectorLoader::loadInternal(
         vector->setNull(vectorIndex, false);
         break;
       }
+      case ColumnType::Timestamp: {
+        auto timestampVector = vector->asFlatVector<Timestamp>();
+        if (value->is<double>()) {
+          timestampVector->set(
+              vectorIndex,
+              convertToVeloxTimestamp(value->get_immutable_view<double>()));
+        } else if (value->is<int64_t>()) {
+          timestampVector->set(
+              vectorIndex,
+              convertToVeloxTimestamp(value->get_immutable_view<int64_t>()));
+        } else if (value->is<std::string>()) {
+          auto stringValue = value->get_immutable_view<std::string>().data();
+          uint64_t encodingId{};
+          auto const timestamp = timestampDict_.ingest_entry(
+              nodeName_, nodeId, stringValue, encodingId);
+          timestampVector->set(vectorIndex, convertToVeloxTimestamp(timestamp));
+        } else {
+          VELOX_FAIL("Unsupported timestamp type");
+        }
+        break;
+      }
       case ColumnType::Array: {
         auto arrayVector = std::dynamic_pointer_cast<ArrayVector>(vector);
         std::string jsonString;