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/ClpColumnHandle.h

@@ -42,6 +42,8 @@ class ClpColumnHandle : public ColumnHandle {
     return columnType_;
   }
 
+  inline static const std::string jsonStringColumnName_ = "__json_string";
+
  private:
   const std::string columnName_;
   const std::string originalColumnName_;

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

@@ -20,7 +20,9 @@
 #include "clp_s/search/EvaluateTimestampIndex.hpp"
 #include "clp_s/search/ast/EmptyExpr.hpp"
 #include "clp_s/search/ast/SearchUtils.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"
 #include "velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h"
 #include "velox/connectors/clp/search_lib/archive/ClpQueryRunner.h"
 
@@ -45,7 +47,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 +72,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 +102,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 +152,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 (ClpColumnHandle::jsonStringColumnName_ == column.name) {
+        jsonStringColumnIndices_.insert(i);
+        continue;
+      }
       std::vector<std::string> descriptorTokens;
       std::string descriptorNamespace;
       if (false ==
@@ -195,7 +203,6 @@ ErrorCode ClpArchiveCursor::loadSplit() {
     return ErrorCode::InternalError;
   }
   projection_->resolve_columns(schemaTree);
-  archiveReader_->set_projection(projection_);
 
   archiveReader_->read_metadata();
 
@@ -237,9 +244,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 +259,25 @@ VectorPtr ClpArchiveCursor::createVectorHelper(
   auto vector = BaseVector::create(vectorType, vectorSize, pool);
   vector->setNulls(allocateNulls(vectorSize, pool, bits::kNull));
 
+  if (jsonStringColumnIndices_.contains(columnIndex_)) {
+    ++columnIndex_;
+    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_[columnIndex_].type;
+  projectedColumnIndex_++;
+  columnIndex_++;
   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,51 @@
+/*
+ * 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");
+  VELOX_CHECK_NULL(
+      hook, "ClpArchiveJsonStringVectorLoader doesn't support ValueHook");
+
+  auto vector = *result;
+  auto* stringVector = vector->asFlatVector<StringView>();
+  for (vector_size_t 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,44 @@
+/*
+ * 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 "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 for the archive format.
+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/search_lib/ir/CMakeLists.txt

@@ -15,10 +15,8 @@ velox_add_library(
   clp-s-ir-search
   STATIC
   ClpIrCursor.cpp
-  ClpIrCursor.h
-  ClpIrUnitHandler.h
-  ClpIrVectorLoader.cpp
-  ClpIrVectorLoader.h)
+  ClpIrJsonStringVectorLoader.cpp
+  ClpIrVectorLoader.cpp)
 
 velox_link_libraries(
   clp-s-ir-search

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

@@ -18,15 +18,18 @@
 #include "clp_s/InputConfig.hpp"
 
 #include "ffi/ir_stream/search/QueryHandler.hpp"
+#include "velox/connectors/clp/ClpColumnHandle.h"
 #include "velox/connectors/clp/search_lib/ir/ClpIrCursor.h"
+#include "velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.h"
 #include "velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h"
 
 using namespace clp_s;
 
 namespace facebook::velox::connector::clp::search_lib {
 
 uint64_t ClpIrCursor::fetchNext(uint64_t numRows) {
-  readerIndex_ = 0;
+  columnIndex_ = 0;
+  projectedColumnIndex_ = 0;
   if (ErrorCode::Success != errorCode_) {
     return 0;
   }
@@ -59,10 +62,10 @@ VectorPtr ClpIrCursor::createVector(
     const TypePtr& vectorType,
     size_t vectorSize) {
   VELOX_CHECK_LE(
-      projectedColumnIdxNodeIdsMap_.size(),
+      projectedColumnIdxNodeIdsMap_.size() + jsonStringColumnIndices_.size(),
       outputColumns_.size(),
-      "Resolved node-id map size ({}) must not exceed projected columns ({})",
-      projectedColumnIdxNodeIdsMap_.size(),
+      "Resolved node-id map size ({}) must not exceed projected columns and json string columns ({})",
+      projectedColumnIdxNodeIdsMap_.size() + jsonStringColumnIndices_.size(),
       outputColumns_.size());
   return createVectorHelper(pool, vectorType, vectorSize);
 }
@@ -120,11 +123,15 @@ ErrorCode ClpIrCursor::loadSplit() {
 }
 
 std::vector<std::pair<std::string, search::ast::literal_type_bitmask_t>>
-ClpIrCursor::splitFieldsToNamesAndTypes() const {
+ClpIrCursor::splitFieldsToNamesAndTypes() {
   auto result = std::vector<
       std::pair<std::string, search::ast::literal_type_bitmask_t>>{};
   for (size_t i{0}; i < outputColumns_.size(); ++i) {
     auto column = outputColumns_[i];
+    if (ClpColumnHandle::jsonStringColumnName_ == column.name) {
+      jsonStringColumnIndices_.insert(i);
+      continue;
+    }
     search::ast::literal_type_bitmask_t literalType;
     switch (column.type) {
       case ColumnType::Array:
@@ -195,18 +202,32 @@ VectorPtr ClpIrCursor::createVectorHelper(
   }
   auto vector = BaseVector::create(vectorType, vectorSize, pool);
   vector->setNulls(allocateNulls(vectorSize, pool, bits::kNull));
+
+  if (jsonStringColumnIndices_.contains(columnIndex_)) {
+    ++columnIndex_;
+    return std::make_shared<LazyVector>(
+        pool,
+        vectorType,
+        vectorSize,
+        std::make_unique<ClpIrJsonStringVectorLoader>(filteredLogEvents_),
+        std::move(vector));
+  }
+
   VELOX_CHECK_LT(
-      readerIndex_, outputColumns_.size(), "Reader index out of bounds");
-  auto projectedColumn = outputColumns_[readerIndex_];
+      columnIndex_,
+      outputColumns_.size(),
+      "Projected column index out of bounds");
+  auto projectedColumn = outputColumns_[columnIndex_];
   auto projectedColumnType = projectedColumn.type;
-  auto it = projectedColumnIdxNodeIdsMap_.find(readerIndex_);
+  auto it = projectedColumnIdxNodeIdsMap_.find(projectedColumnIndex_);
   std::vector<::clp::ffi::SchemaTree::Node::id_t> projectedColumnNodeIds{};
   bool isResolved =
       it != projectedColumnIdxNodeIdsMap_.end() && !it->second.empty();
   if (isResolved) {
     projectedColumnNodeIds = it->second;
   }
-  readerIndex_++;
+  projectedColumnIndex_++;
+  columnIndex_++;
   return std::make_shared<LazyVector>(
       pool,
       vectorType,

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

@@ -16,6 +16,8 @@
 
 #pragma once
 
+#include <unordered_set>
+
 #include "clp/streaming_compression/zstd/Decompressor.hpp"
 #include "ffi/ir_stream/Deserializer.hpp"
 #include "velox/connectors/clp/search_lib/BaseClpCursor.h"
@@ -85,11 +87,13 @@ class ClpIrCursor final : public BaseClpCursor {
       irReaderZstdWrapper_{nullptr};
   std::unordered_map<size_t, std::vector<::clp::ffi::SchemaTree::Node::id_t>>
       projectedColumnIdxNodeIdsMap_;
-  size_t readerIndex_{0};
+  std::unordered_set<size_t> jsonStringColumnIndices_;
+  size_t columnIndex_{0};
+  size_t projectedColumnIndex_{0};
 
   std::vector<
       std::pair<std::string, clp_s::search::ast::literal_type_bitmask_t>>
-  splitFieldsToNamesAndTypes() const;
+  splitFieldsToNamesAndTypes();
 
   ystdlib::error_handling::Result<void> deserialize(uint64_t numRows);