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

velox/connectors/clp/search_lib/CMakeLists.txt

@@ -19,7 +19,8 @@ velox_add_library(
   ClpPackageS3AuthProvider.cpp
   ClpPackageS3AuthProvider.h
   ClpS3AuthProviderBase.cpp
-  ClpS3AuthProviderBase.h)
+  ClpS3AuthProviderBase.h
+  ClpTimestampsUtils.h)
 
 add_subdirectory(archive)
 add_subdirectory(ir)

velox/connectors/clp/search_lib/ClpTimestampsUtils.h

@@ -0,0 +1,120 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include "velox/type/Timestamp.h"
+
+namespace facebook::velox::connector::clp::search_lib {
+
+enum class InputTimestampPrecision : 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) -> InputTimestampPrecision {
+  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 InputTimestampPrecision::Nanoseconds;
+  } else if (absTimestamp > kEpochMicroseconds1971) {
+    return InputTimestampPrecision::Microseconds;
+  } else if (absTimestamp > kEpochMilliseconds1971) {
+    return InputTimestampPrecision::Milliseconds;
+  } else {
+    return InputTimestampPrecision::Seconds;
+  }
+}
+
+/// Converts a double value into a Velox timestamp.
+///
+/// @param timestamp the input timestamp as a double
+/// @return the corresponding Velox timestamp
+inline auto convertToVeloxTimestamp(double timestamp) -> Timestamp {
+  switch (estimatePrecision(timestamp)) {
+    case InputTimestampPrecision::Nanoseconds:
+      timestamp /= Timestamp::kNanosInSecond;
+      break;
+    case InputTimestampPrecision::Microseconds:
+      timestamp /= Timestamp::kMicrosecondsInSecond;
+      break;
+    case InputTimestampPrecision::Milliseconds:
+      timestamp /= Timestamp::kMillisecondsInSecond;
+      break;
+    case InputTimestampPrecision::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));
+}
+
+/// Converts an integer value into a Velox timestamp.
+///
+/// @param timestamp the input timestamp as an integer
+/// @return the corresponding Velox timestamp
+inline auto convertToVeloxTimestamp(int64_t timestamp) -> Timestamp {
+  int64_t precisionDifference{Timestamp::kNanosInSecond};
+  switch (estimatePrecision(timestamp)) {
+    case InputTimestampPrecision::Nanoseconds:
+      break;
+    case InputTimestampPrecision::Microseconds:
+      precisionDifference =
+          Timestamp::kNanosInSecond / Timestamp::kNanosecondsInMicrosecond;
+      break;
+    case InputTimestampPrecision::Milliseconds:
+      precisionDifference =
+          Timestamp::kNanosInSecond / Timestamp::kNanosecondsInMillisecond;
+      break;
+    case InputTimestampPrecision::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 facebook::velox::connector::clp::search_lib

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

@@ -21,106 +21,14 @@
 #include "clp_s/ColumnReader.hpp"
 #include "clp_s/SchemaTree.hpp"
 #include "velox/connectors/clp/search_lib/BaseClpCursor.h"
+#include "velox/connectors/clp/search_lib/ClpTimestampsUtils.h"
 #include "velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.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
-
 ClpArchiveVectorLoader::ClpArchiveVectorLoader(
     clp_s::BaseColumnReader* columnReader,
     ColumnType nodeType,

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

@@ -29,5 +29,6 @@ velox_link_libraries(
     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

@@ -18,6 +18,7 @@
 #include "clp_s/InputConfig.hpp"
 
 #include "ffi/ir_stream/search/QueryHandler.hpp"
+#include "velox/connectors/clp/search_lib/ClpTimestampsUtils.h"
 #include "velox/connectors/clp/search_lib/ir/ClpIrCursor.h"
 #include "velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h"
 
@@ -59,10 +60,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(),
+      "Resolved node-id map size ({}) must not exceed projected columns ({})",
+      projectedColumnIdxNodeIdsMap_.size(),
       outputColumns_.size());
   return createVectorHelper(pool, vectorType, vectorSize);
 }
@@ -135,7 +136,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 +191,24 @@ 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_);
+  std::vector<::clp::ffi::SchemaTree::Node::id_t> projectedColumnNodeIds{};
+  bool isResolved =
+      it != projectedColumnIdxNodeIdsMap_.end() && !it->second.empty();
   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()),
+          std::move(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

@@ -15,7 +15,9 @@
  */
 
 #include "velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h"
+
 #include "velox/connectors/clp/search_lib/BaseClpCursor.h"
+#include "velox/connectors/clp/search_lib/ClpTimestampsUtils.h"
 
 namespace facebook::velox::connector::clp::search_lib {
 
@@ -33,12 +35,20 @@ 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()) {
+    auto valueIt = userGenNodeIdValueMap.end();
+    ::clp::ffi::SchemaTree::Node::id_t nodeId{};
+    for (auto const candidateNodeId : nodeIds_) {
+      valueIt = userGenNodeIdValueMap.find(candidateNodeId);
+      if (valueIt != userGenNodeIdValueMap.end()) {
+        nodeId = candidateNodeId;
+        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 +99,21 @@ 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 {
+          VELOX_FAIL("Unsupported timestamp type");
+        }
+        break;
+      }
       case ColumnType::Array: {
         auto arrayVector = std::dynamic_pointer_cast<ArrayVector>(vector);
         std::string jsonString;