feat: Add support for CLP JSON string serialization.

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 67276c09acbd48dd502454288f40072c44628726)
+  GIT_TAG 9e991ab49681faff0aea259a37b489c3f416e06e)
 
 set(CLP_BUILD_CLP_REGEX_UTILS
     OFF

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

@@ -15,11 +15,9 @@ velox_add_library(
   clp-s-archive-search
   STATIC
   ClpArchiveCursor.cpp
-  ClpArchiveCursor.h
+  ClpArchiveJsonStringVectorLoader.cpp
   ClpArchiveVectorLoader.cpp
-  ClpArchiveVectorLoader.h
-  ClpQueryRunner.cpp
-  ClpQueryRunner.h)
+  ClpQueryRunner.cpp)
 
 velox_link_libraries(
   clp-s-archive-search

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

@@ -21,6 +21,7 @@
 #include "clp_s/search/ast/EmptyExpr.hpp"
 #include "clp_s/search/ast/SearchUtils.hpp"
 #include "velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h"
+#include "velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h"
 #include "velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h"
 #include "velox/connectors/clp/search_lib/archive/ClpQueryRunner.h"
 
@@ -45,7 +46,8 @@ ClpArchiveCursor::~ClpArchiveCursor() {
 
 uint64_t ClpArchiveCursor::fetchNext(uint64_t numRows) {
   filteredRowIndices_->clear();
-  readerIndex_ = 0;
+  projectedColumnIndex_ = 0;
+  columnIndex_ = 0;
 
   if (ErrorCode::Success != errorCode_) {
     return 0;
@@ -69,9 +71,9 @@ uint64_t ClpArchiveCursor::fetchNext(uint64_t numRows) {
         continue;
       }
 
-      auto& reader =
-          archiveReader_->read_schema_table(currentSchemaId_, false, false);
-      reader.initialize_filter_with_column_map(queryRunner_.get());
+      schemaReader_ =
+          &archiveReader_->read_schema_table(currentSchemaId_, false, false);
+      schemaReader_->initialize_filter_with_column_map(queryRunner_.get());
 
       errorCode_ = ErrorCode::Success;
       currentSchemaTableLoaded_ = true;
@@ -99,7 +101,7 @@ VectorPtr ClpArchiveCursor::createVector(
     size_t vectorSize) {
   auto projectedColumns = getProjectedColumns();
   VELOX_CHECK_EQ(
-      projectedColumns.size(),
+      projectedColumns.size() + jsonStringColumnIndices_.size(),
       outputColumns_.size(),
       "Projected columns size {} does not match fields size {}",
       projectedColumns.size(),
@@ -149,7 +151,12 @@ ErrorCode ClpArchiveCursor::loadSplit() {
   projection_ = std::make_shared<Projection>(
       outputColumns_.empty() ? ReturnAllColumns : ReturnSelectedColumns, true);
   try {
-    for (auto const& column : outputColumns_) {
+    for (size_t i = 0; i < outputColumns_.size(); i++) {
+      auto const& column = outputColumns_[i];
+      if ("__json_string" == column.name) {
+        jsonStringColumnIndices_.insert(i);
+        continue;
+      }
       std::vector<std::string> descriptorTokens;
       std::string descriptorNamespace;
       if (false ==
@@ -195,7 +202,6 @@ ErrorCode ClpArchiveCursor::loadSplit() {
     return ErrorCode::InternalError;
   }
   projection_->resolve_columns(schemaTree);
-  archiveReader_->set_projection(projection_);
 
   archiveReader_->read_metadata();
 
@@ -237,9 +243,12 @@ VectorPtr ClpArchiveCursor::createVectorHelper(
     size_t vectorSize,
     const std::vector<clp_s::BaseColumnReader*>& projectedColumns) {
   if (vectorType->kind() == TypeKind::ROW) {
-    std::vector<VectorPtr> children;
     auto& rowType = vectorType->as<TypeKind::ROW>();
-    for (uint32_t i = 0; i < rowType.size(); ++i) {
+    const uint32_t numChildren = rowType.size();
+    std::vector<VectorPtr> children;
+    children.reserve(numChildren);
+
+    for (uint32_t i = 0; i < numChildren; ++i) {
       children.push_back(createVectorHelper(
           pool, rowType.childAt(i), vectorSize, projectedColumns));
     }
@@ -249,11 +258,25 @@ VectorPtr ClpArchiveCursor::createVectorHelper(
   auto vector = BaseVector::create(vectorType, vectorSize, pool);
   vector->setNulls(allocateNulls(vectorSize, pool, bits::kNull));
 
+  const bool isJsonString = jsonStringColumnIndices_.contains(columnIndex_);
+  ++columnIndex_;
+  if (isJsonString) {
+    return std::make_shared<LazyVector>(
+        pool,
+        vectorType,
+        vectorSize,
+        std::make_unique<ClpArchiveJsonStringVectorLoader>(
+            schemaReader_, filteredRowIndices_),
+        std::move(vector));
+  }
+
   VELOX_CHECK_LT(
-      readerIndex_, projectedColumns.size(), "Reader index out of bounds");
-  auto projectedColumn = projectedColumns[readerIndex_];
-  auto projectedType = outputColumns_[readerIndex_].type;
-  readerIndex_++;
+      projectedColumnIndex_,
+      projectedColumns.size(),
+      "Projected column index out of bounds");
+  auto* projectedColumn = projectedColumns[projectedColumnIndex_];
+  auto projectedType = outputColumns_[projectedColumnIndex_].type;
+  projectedColumnIndex_++;
   return std::make_shared<LazyVector>(
       pool,
       vectorType,

velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h

@@ -16,10 +16,13 @@
 
 #pragma once
 
+#include <unordered_set>
+
 #include "velox/connectors/clp/search_lib/BaseClpCursor.h"
 
 namespace clp_s {
 class ArchiveReader;
+class SchemaReader;
 } // namespace clp_s
 
 namespace clp_s::search {
@@ -54,14 +57,17 @@ class ClpArchiveCursor final : public BaseClpCursor {
 
  private:
   std::shared_ptr<clp_s::ArchiveReader> archiveReader_;
+  clp_s::SchemaReader* schemaReader_;
   int32_t currentSchemaId_{-1};
   size_t currentSchemaIndex_{0};
   bool currentSchemaTableLoaded_{false};
   std::shared_ptr<std::vector<uint64_t>> filteredRowIndices_;
   std::vector<int32_t> matchedSchemas_;
   std::shared_ptr<clp_s::search::Projection> projection_;
   std::shared_ptr<ClpQueryRunner> queryRunner_;
-  size_t readerIndex_{0};
+  size_t columnIndex_{0};
+  size_t projectedColumnIndex_{0};
+  std::unordered_set<size_t> jsonStringColumnIndices_;
   std::shared_ptr<clp_s::search::SchemaMatch> schemaMatch_;
 
   const std::vector<clp_s::BaseColumnReader*>& getProjectedColumns() const;

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

@@ -0,0 +1,49 @@
+/*
+ * 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.
+ */
+
+#include <cstdint>
+
+#include "clp_s/ColumnReader.hpp"
+#include "velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h"
+#include "velox/vector/ComplexVector.h"
+
+namespace facebook::velox::connector::clp::search_lib {
+
+ClpArchiveJsonStringVectorLoader::ClpArchiveJsonStringVectorLoader(
+    clp_s::SchemaReader* schemaReader,
+    const std::shared_ptr<std::vector<uint64_t>>& filteredRowIndices)
+    : schemaReader_(schemaReader), filteredRowIndices_(filteredRowIndices) {
+  schemaReader_->initialize_serializer();
+}
+
+void ClpArchiveJsonStringVectorLoader::loadInternal(
+    RowSet rows,
+    ValueHook* hook,
+    vector_size_t resultSize,
+    VectorPtr* result) {
+  VELOX_CHECK_NOT_NULL(result, "result vector must not be null");
+
+  auto vector = *result;
+  auto* stringVector = vector->asFlatVector<StringView>();
+  for (int const vectorIndex : rows) {
+    auto messageIndex = filteredRowIndices_->at(vectorIndex);
+    auto jsonString = schemaReader_->generate_json_string(messageIndex);
+    stringVector->set(vectorIndex, StringView(jsonString));
+    stringVector->setNull(vectorIndex, false);
+  }
+}
+
+} // namespace facebook::velox::connector::clp::search_lib

velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h

@@ -0,0 +1,46 @@
+/*
+ * 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 <simdjson.h>
+
+#include "clp_s/SchemaReader.hpp"
+#include "velox/vector/FlatVector.h"
+#include "velox/vector/LazyVector.h"
+
+namespace facebook::velox::connector::clp::search_lib {
+
+/// A custom Velox VectorLoader that populates Velox vectors with serialized
+/// JSON string from CLP column readers.
+class ClpArchiveJsonStringVectorLoader : public VectorLoader {
+ public:
+  ClpArchiveJsonStringVectorLoader(
+      clp_s::SchemaReader* schemaReader,
+      const std::shared_ptr<std::vector<uint64_t>>& filteredRowIndices);
+
+ private:
+  clp_s::SchemaReader* schemaReader_;
+  std::shared_ptr<std::vector<uint64_t>> filteredRowIndices_;
+
+  void loadInternal(
+      RowSet rows,
+      ValueHook* hook,
+      vector_size_t resultSize,
+      VectorPtr* result) override;
+};
+
+} // namespace facebook::velox::connector::clp::search_lib

velox/connectors/clp/tests/ClpConnectorTest.cpp

@@ -207,6 +207,59 @@ TEST_F(ClpConnectorTest, test1Pushdown) {
   test::assertEqualVectors(expected, irOutput);
 }
 
+TEST_F(ClpConnectorTest, test1JsonString) {
+  const std::shared_ptr<std::string> kqlQuery = nullptr;
+  auto plan = PlanBuilder()
+                  .startTableScan()
+                  .outputType(
+                      ROW({"requestId", "__json_string", "method"},
+                          {VARCHAR(), VARCHAR(), VARCHAR()}))
+                  .tableHandle(std::make_shared<ClpTableHandle>(
+                      kClpConnectorId, "test_1"))
+                  .assignments({
+                      {"requestId",
+                       std::make_shared<ClpColumnHandle>(
+                           "requestId", "requestId", VARCHAR())},
+                      {"__json_string",
+                       std::make_shared<ClpColumnHandle>(
+                           "__json_string", "__json_string", VARCHAR())},
+                      {"method",
+                       std::make_shared<ClpColumnHandle>(
+                           "method", "method", VARCHAR())},
+                  })
+                  .endTableScan()
+                  .filter("method = 'GET'")
+                  .planNode();
+
+  auto output = getResults(
+      plan,
+      {makeClpSplit(
+          getExampleFilePath("test_1.clps"),
+          ClpConnectorSplit::SplitType::kArchive,
+          kqlQuery)});
+  auto expected = makeRowVector(
+      {// requestId
+       makeFlatVector<StringView>(
+           {"req-100", "req-105", "req-107", "req-109", "req-102"}),
+       // __json_string
+       makeFlatVector<StringView>({
+           R"({"timestamp":"2025-04-30T08:45:00Z","requestId":"req-100","userId":"user201","method":"GET","path":"/api/users/1","responseTimeMs":25,"status":200})",
+           R"({"timestamp":"2025-04-30T08:45:25Z","requestId":"req-105","userId":"user204","method":"GET","path":"/api/dashboard","responseTimeMs":155,"status":200})",
+           R"({"timestamp":"2025-04-30T08:45:35Z","requestId":"req-107","userId":"user202","method":"GET","path":"/api/users/2/details","responseTimeMs":41,"status":200})",
+           R"({"timestamp":"2025-04-30T08:45:45Z","requestId":"req-109","userId":"user203","method":"GET","path":"/api/products?category=books","responseTimeMs":88,"status":200})",
+           R"({"timestamp":"2025-04-30T08:45:10Z","requestId":"req-102","method":"GET","path":"/public/products","responseTimeMs":18,"status":200,"userId":null})",
+       }),
+       // method
+       makeFlatVector<StringView>({
+           "GET",
+           "GET",
+           "GET",
+           "GET",
+           "GET",
+       })});
+  test::assertEqualVectors(expected, output);
+}
+
 TEST_F(ClpConnectorTest, test2NoPushdown) {
   const std::shared_ptr<std::string> kqlQuery = nullptr;
   auto plan =
@@ -384,6 +437,43 @@ TEST_F(ClpConnectorTest, test2Hybrid) {
   test::assertEqualVectors(expected, irOutput);
 }
 
+TEST_F(ClpConnectorTest, test2JsonSring) {
+  auto kqlQuery = std::make_shared<std::string>(
+      "(event.severity: \"WARNING\" OR event.severity: \"ERROR\") AND "
+      "((event.type: \"network\" AND event.subtype: \"connection\") OR "
+      "(event.type: \"storage\" AND event.subtype: \"disk*\"))");
+  auto plan = PlanBuilder()
+                  .startTableScan()
+                  .outputType(ROW(
+                      {"timestamp", "__json_string"}, {TIMESTAMP(), VARCHAR()}))
+                  .tableHandle(std::make_shared<ClpTableHandle>(
+                      kClpConnectorId, "test_2"))
+                  .assignments(
+                      {{"timestamp",
+                        std::make_shared<ClpColumnHandle>(
+                            "timestamp", "timestamp", TIMESTAMP())},
+                       {"__json_string",
+                        std::make_shared<ClpColumnHandle>(
+                            "__json_string", "__json_string", VARCHAR())}})
+                  .endTableScan()
+                  .planNode();
+
+  auto output = getResults(
+      plan,
+      {makeClpSplit(
+          getExampleFilePath("test_2.clps"),
+          ClpConnectorSplit::SplitType::kArchive,
+          kqlQuery)});
+  auto expected = makeRowVector(
+      {// timestamp
+       makeFlatVector<Timestamp>(
+           {Timestamp(kTestTimestampSeconds, kTestTimestampNanoseconds)}),
+       // __json_string
+       makeFlatVector<StringView>(
+           {R"({"timestamp":"2025-04-30T08:50:05Z","event":{"type":"storage","subtype":"disk_usage","severity":"WARNING","tags":["filesystem", "monitoring"],"details":{"mount":"/var/log","usage":{"percent":92}}}})"})});
+  test::assertEqualVectors(expected, output);
+}
+
 TEST_F(ClpConnectorTest, test3TimestampMarshalling) {
   const std::shared_ptr<std::string> kqlQuery = nullptr;
   auto plan = PlanBuilder(pool_.get())