feat: Add support for searching and marshalling the new Timestamp column type. (#54)

Author: gibber9809

CMake/resolve_dependency_modules/clp.cmake

@@ -16,7 +16,7 @@ include_guard(GLOBAL)
 FetchContent_Declare(
   clp
   GIT_REPOSITORY https://github.com/y-scope/clp.git
-  GIT_TAG f82e6114160a6addd4727259906bcf621ac9912c
+  GIT_TAG a91e5f71f0715d7d6b3ea7c177e1b39b3e6a24a6
 )
 
 set(CLP_BUILD_CLP_REGEX_UTILS OFF CACHE BOOL "Build CLP regex utils")

velox/connectors/clp/search_lib/ClpTimestampsUtils.h

@@ -16,6 +16,7 @@
 
 #pragma once
 
+#include "clp_s/Defs.hpp"
 #include "velox/type/Timestamp.h"
 
 namespace facebook::velox::connector::clp::search_lib {
@@ -121,4 +122,19 @@ inline auto convertToVeloxTimestamp(int64_t timestamp) -> Timestamp {
   return Timestamp(seconds, static_cast<uint64_t>(nanoseconds));
 }
 
+/// Converts a nanosecond precision epochtime_t into a Velox timestamp.
+///
+/// @param timestamp the input timestamp as an integer
+/// @return the corresponding Velox timestamp
+inline auto convertNanosecondEpochToVeloxTimestamp(clp_s::epochtime_t timestamp)
+    -> Timestamp {
+  int64_t seconds{timestamp / Timestamp::kNanosInSecond};
+  int64_t nanoseconds{timestamp % Timestamp::kNanosInSecond};
+  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/ClpArchiveCursor.cpp

@@ -17,9 +17,12 @@
 #include <glog/logging.h>
 
 #include "clp_s/ArchiveReader.hpp"
+#include "clp_s/SingleFileArchiveDefs.hpp"
 #include "clp_s/search/EvaluateTimestampIndex.hpp"
 #include "clp_s/search/ast/EmptyExpr.hpp"
 #include "clp_s/search/ast/SearchUtils.hpp"
+#include "clp_s/search/ast/SetTimestampLiteralPrecision.hpp"
+#include "clp_s/search/ast/TimestampLiteral.hpp"
 #include "velox/connectors/clp/ClpColumnHandle.h"
 #include "velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h"
 #include "velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h"
@@ -136,6 +139,14 @@ ErrorCode ClpArchiveCursor::loadSplit() {
   auto schemaTree = archiveReader_->get_schema_tree();
   auto schemaMap = archiveReader_->get_schema_map();
 
+  auto const defaultTimestampPrecision{
+      archiveReader_->has_deprecated_timestamp_format()
+          ? TimestampLiteral::Precision::Milliseconds
+          : TimestampLiteral::Precision::Nanoseconds};
+  SetTimestampLiteralPrecision timestampPrecisionPass{
+      defaultTimestampPrecision};
+  expr_ = timestampPrecisionPass.run(expr_);
+
   EvaluateTimestampIndex timestampIndex(timestampDict);
   if (clp_s::EvaluatedValue::False == timestampIndex.run(expr_)) {
     VLOG(2) << "No matching timestamp ranges for query '" << query_ << "'";

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

@@ -73,6 +73,7 @@ void ClpArchiveVectorLoader::populateTimestampData(
     case clp_s::NodeType::DictionaryFloat:
     case clp_s::NodeType::Integer:
     case clp_s::NodeType::DeprecatedDateString:
+    case clp_s::NodeType::Timestamp:
       supportedNodeType = true;
       break;
     default:
@@ -88,7 +89,13 @@ void ClpArchiveVectorLoader::populateTimestampData(
   for (int vectorIndex : rows) {
     auto messageIndex = filteredRowIndices_->at(vectorIndex);
 
-    if (clp_s::NodeType::Float == Type) {
+    if (clp_s::NodeType::Timestamp == Type) {
+      auto reader{static_cast<clp_s::TimestampColumnReader*>(columnReader_)};
+      vector->set(
+          vectorIndex,
+          convertNanosecondEpochToVeloxTimestamp(
+              reader->get_encoded_time(messageIndex)));
+    } else if (clp_s::NodeType::Float == Type) {
       auto reader = static_cast<clp_s::FloatColumnReader*>(columnReader_);
       vector->set(
           vectorIndex,
@@ -211,7 +218,12 @@ void ClpArchiveVectorLoader::loadInternal(
     }
     case ColumnType::Timestamp: {
       auto timestampVector = vector->asFlatVector<Timestamp>();
-      if (nullptr != dynamic_cast<clp_s::Int64ColumnReader*>(columnReader_)) {
+      if (nullptr !=
+          dynamic_cast<clp_s::TimestampColumnReader*>(columnReader_)) {
+        populateTimestampData<clp_s::NodeType::Timestamp>(
+            rows, timestampVector);
+      } else if (
+          nullptr != dynamic_cast<clp_s::Int64ColumnReader*>(columnReader_)) {
         populateTimestampData<clp_s::NodeType::Integer>(rows, timestampVector);
       } else if (
           nullptr !=
@@ -256,10 +268,22 @@ template void ClpArchiveVectorLoader::populateData<std::string>(
     RowSet rows,
     FlatVector<StringView>* vector);
 template void
+ClpArchiveVectorLoader::populateTimestampData<clp_s::NodeType::Timestamp>(
+    RowSet rows,
+    FlatVector<facebook::velox::Timestamp>* vector);
+template void
 ClpArchiveVectorLoader::populateTimestampData<clp_s::NodeType::Float>(
     RowSet rows,
     FlatVector<facebook::velox::Timestamp>* vector);
 template void
+ClpArchiveVectorLoader::populateTimestampData<clp_s::NodeType::FormattedFloat>(
+    RowSet rows,
+    FlatVector<facebook::velox::Timestamp>* vector);
+template void
+ClpArchiveVectorLoader::populateTimestampData<clp_s::NodeType::DictionaryFloat>(
+    RowSet rows,
+    FlatVector<facebook::velox::Timestamp>* vector);
+template void
 ClpArchiveVectorLoader::populateTimestampData<clp_s::NodeType::Integer>(
     RowSet rows,
     FlatVector<facebook::velox::Timestamp>* vector);

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

@@ -16,6 +16,8 @@
 
 #include "clp_s/ColumnReader.hpp"
 #include "clp_s/InputConfig.hpp"
+#include "clp_s/search/ast/SetTimestampLiteralPrecision.hpp"
+#include "clp_s/search/ast/TimestampLiteral.hpp"
 
 #include "ffi/ir_stream/search/QueryHandler.hpp"
 #include "velox/connectors/clp/ClpColumnHandle.h"
@@ -75,6 +77,10 @@ ErrorCode ClpIrCursor::loadSplit() {
       ? NetworkAuthOption{.method = AuthMethod::None}
       : NetworkAuthOption{.method = AuthMethod::S3PresignedUrlV4};
 
+  search::ast::SetTimestampLiteralPrecision timestampPrecisionPass{
+      search::ast::TimestampLiteral::Precision::Milliseconds};
+  expr_ = timestampPrecisionPass.run(expr_);
+
   auto projections = splitFieldsToNamesAndTypes();
   auto queryHandlerResult{QueryHandlerType::create(
       projectionResolutionCallback_,

velox/connectors/clp/tests/ClpConnectorTest.cpp

@@ -580,6 +580,10 @@ TEST_F(ClpConnectorTest, test4IrTimestampNoPushdown) {
 TEST_F(ClpConnectorTest, test4IrTimestampPushdown) {
   // Only the second event meet the condition, the first event is a date string
   // which is not supported yet so the value will be NULL.
+  // This test can not use the `timestamp()` literal, since the integer
+  // timestamps are in microsecond precision, and we currently assume all IR
+  // timestamps are millisecond precision when comparing against timestamp
+  // literals.
   const std::shared_ptr<std::string> kqlQuery =
       std::make_shared<std::string>("(timestamp < 1756003005000000)");
   auto plan =
@@ -661,11 +665,9 @@ TEST_F(ClpConnectorTest, test5FloatTimestampNoPushdown) {
 
 TEST_F(ClpConnectorTest, test5FloatTimestampPushdown) {
   // Test filtering rows with a timestamp parsed from a date string and floats
-  // in various formats. Because KQL doesn’t automatically interpret the unit of
-  // the timestamp, the returned result differs slightly from the one without
-  // pushdown.
+  // in various formats.
   const std::shared_ptr<std::string> kqlQuery = std::make_shared<std::string>(
-      "(timestamp < 1746003005.127 and timestamp >= 1746003005.124)");
+      R"(timestamp < timestamp("1746003070000", "\L") and timestamp >= timestamp("1746003005124", "\L"))");
   auto plan =
       PlanBuilder(pool_.get())
           .startTableScan()
@@ -695,12 +697,67 @@ TEST_F(ClpConnectorTest, test5FloatTimestampPushdown) {
            {Timestamp(1746003005, 124000000),
             Timestamp(1746003005, 124100000),
             Timestamp(1746003005, 125000000),
-            Timestamp(1746003005, 126000000)}),
+            Timestamp(1746003005, 126000000),
+            Timestamp(1746003005, 127000000),
+            Timestamp(1746003060, 0),
+            Timestamp(1746003065, 0)}),
        makeFlatVector<double>(
-           {1.234567891234500E9,
+           {1.2345678912345E9,
             1E16,
             1.234567891234567E9,
-            1.234567891234567E9})});
+            1.234567891234567E9,
+            -1.234567891234567E-9,
+            1234567891.234567,
+            -1234567891.234567})});
+  test::assertEqualVectors(expected, output);
+}
+
+TEST_F(ClpConnectorTest, test5NewTimestampFormatFloatTimestampPushdown) {
+  // Test filtering rows with a timestamp parsed from a date string and floats
+  // in various formats.
+  const std::shared_ptr<std::string> kqlQuery = std::make_shared<std::string>(
+      R"(timestamp < timestamp("1746003070000", "\L") and timestamp >= timestamp("1746003005124", "\L"))");
+  auto plan =
+      PlanBuilder(pool_.get())
+          .startTableScan()
+          .outputType(ROW({"timestamp", "floatValue"}, {TIMESTAMP(), DOUBLE()}))
+          .tableHandle(
+              std::make_shared<ClpTableHandle>(kClpConnectorId, "test_5"))
+          .assignments(
+              {{"timestamp",
+                std::make_shared<ClpColumnHandle>(
+                    "timestamp", "timestamp", TIMESTAMP())},
+               {"floatValue",
+                std::make_shared<ClpColumnHandle>(
+                    "floatValue", "floatValue", DOUBLE())}})
+          .endTableScan()
+          .orderBy({"\"timestamp\" ASC"}, false)
+          .planNode();
+
+  auto output = getResults(
+      plan,
+      {makeClpSplit(
+          getExampleFilePath("test_5.v0.5.0.clps"),
+          ClpConnectorSplit::SplitType::kArchive,
+          kqlQuery)});
+  auto expected = makeRowVector(
+      {// timestamp
+       makeFlatVector<Timestamp>(
+           {Timestamp(1746003005, 124000000),
+            Timestamp(1746003005, 124100000),
+            Timestamp(1746003005, 125000000),
+            Timestamp(1746003005, 126000000),
+            Timestamp(1746003005, 127000001),
+            Timestamp(1746003060, 0),
+            Timestamp(1746003065, 0)}),
+       makeFlatVector<double>(
+           {1.2345678912345E9,
+            1E16,
+            1.234567891234567E9,
+            1.234567891234567E9,
+            -1.234567891234567E-9,
+            1234567891.234567,
+            -1234567891.234567})});
   test::assertEqualVectors(expected, output);
 }