diff --git a/CMake/resolve_dependency_modules/clp.cmake b/CMake/resolve_dependency_modules/clp.cmake index 0703fdcc735..e94317115db 100644 --- a/CMake/resolve_dependency_modules/clp.cmake +++ b/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 diff --git a/velox/connectors/clp/ClpColumnHandle.h b/velox/connectors/clp/ClpColumnHandle.h index e9b7a135291..d69efdd1749 100644 --- a/velox/connectors/clp/ClpColumnHandle.h +++ b/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_; diff --git a/velox/connectors/clp/search_lib/archive/CMakeLists.txt b/velox/connectors/clp/search_lib/archive/CMakeLists.txt index 9340e22ffbb..d1739d9e2ef 100644 --- a/velox/connectors/clp/search_lib/archive/CMakeLists.txt +++ b/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 diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp index 115cc20bc61..88c4f0b40c7 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp +++ b/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( 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 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& projectedColumns) { if (vectorType->kind() == TypeKind::ROW) { - std::vector children; auto& rowType = vectorType->as(); - for (uint32_t i = 0; i < rowType.size(); ++i) { + const uint32_t numChildren = rowType.size(); + std::vector 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( + pool, + vectorType, + vectorSize, + std::make_unique( + 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( pool, vectorType, diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h index 09bc042d740..ec3e80f5024 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h @@ -16,10 +16,13 @@ #pragma once +#include + #include "velox/connectors/clp/search_lib/BaseClpCursor.h" namespace clp_s { class ArchiveReader; +class SchemaReader; } // namespace clp_s namespace clp_s::search { @@ -54,6 +57,7 @@ class ClpArchiveCursor final : public BaseClpCursor { private: std::shared_ptr archiveReader_; + clp_s::SchemaReader* schemaReader_; int32_t currentSchemaId_{-1}; size_t currentSchemaIndex_{0}; bool currentSchemaTableLoaded_{false}; @@ -61,7 +65,9 @@ class ClpArchiveCursor final : public BaseClpCursor { std::vector matchedSchemas_; std::shared_ptr projection_; std::shared_ptr queryRunner_; - size_t readerIndex_{0}; + size_t columnIndex_{0}; + size_t projectedColumnIndex_{0}; + std::unordered_set jsonStringColumnIndices_; std::shared_ptr schemaMatch_; const std::vector& getProjectedColumns() const; diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.cpp b/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.cpp new file mode 100644 index 00000000000..5ba9dbb467d --- /dev/null +++ b/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 + +#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>& 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(); + 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 diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h b/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h new file mode 100644 index 00000000000..21aa6a87377 --- /dev/null +++ b/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>& filteredRowIndices); + + private: + clp_s::SchemaReader* schemaReader_; + std::shared_ptr> filteredRowIndices_; + + void loadInternal( + RowSet rows, + ValueHook* hook, + vector_size_t resultSize, + VectorPtr* result) override; +}; + +} // namespace facebook::velox::connector::clp::search_lib diff --git a/velox/connectors/clp/search_lib/ir/CMakeLists.txt b/velox/connectors/clp/search_lib/ir/CMakeLists.txt index 1dce25d1af6..568492f385a 100644 --- a/velox/connectors/clp/search_lib/ir/CMakeLists.txt +++ b/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 diff --git a/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp b/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp index 542649e90fe..f26b1cdd810 100644 --- a/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp +++ b/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp @@ -18,7 +18,9 @@ #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; @@ -26,7 +28,8 @@ 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> -ClpIrCursor::splitFieldsToNamesAndTypes() const { +ClpIrCursor::splitFieldsToNamesAndTypes() { auto result = std::vector< std::pair>{}; 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( + pool, + vectorType, + vectorSize, + std::make_unique(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( pool, vectorType, diff --git a/velox/connectors/clp/search_lib/ir/ClpIrCursor.h b/velox/connectors/clp/search_lib/ir/ClpIrCursor.h index 8fd65e28269..7c8af4d053f 100644 --- a/velox/connectors/clp/search_lib/ir/ClpIrCursor.h +++ b/velox/connectors/clp/search_lib/ir/ClpIrCursor.h @@ -16,6 +16,8 @@ #pragma once +#include + #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> projectedColumnIdxNodeIdsMap_; - size_t readerIndex_{0}; + std::unordered_set jsonStringColumnIndices_; + size_t columnIndex_{0}; + size_t projectedColumnIndex_{0}; std::vector< std::pair> - splitFieldsToNamesAndTypes() const; + splitFieldsToNamesAndTypes(); ystdlib::error_handling::Result deserialize(uint64_t numRows); diff --git a/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.cpp b/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.cpp new file mode 100644 index 00000000000..650b4806bb7 --- /dev/null +++ b/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.cpp @@ -0,0 +1,52 @@ +/* + * 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 "velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.h" + +#include "velox/connectors/clp/search_lib/BaseClpCursor.h" +#include "velox/connectors/clp/search_lib/ClpTimestampsUtils.h" + +namespace facebook::velox::connector::clp::search_lib { + +void ClpIrJsonStringVectorLoader::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, "ClpIrJsonStringVectorLoader doesn't support ValueHook"); + + auto vector = *result; + auto* stringVector = vector->asFlatVector(); + for (vector_size_t const vectorIndex : rows) { + const auto& logEvent = filteredLogEvents_->at(vectorIndex); + auto serializedResult = logEvent->serialize_to_json(); + if (serializedResult.has_error()) { + auto error = serializedResult.error(); + VELOX_FAIL( + "Cannot serialize IR to JSON. {}: {}", + error.category().name(), + error.message()); + } + + std::string const jsonString = serializedResult.value().second.dump(); + stringVector->set(vectorIndex, StringView(jsonString)); + stringVector->setNull(vectorIndex, false); + } +} + +} // namespace facebook::velox::connector::clp::search_lib diff --git a/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.h b/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.h new file mode 100644 index 00000000000..eb3ff7b4a27 --- /dev/null +++ b/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.h @@ -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. + */ + +#pragma once + +#include "ffi/ir_stream/Deserializer.hpp" +#include "velox/connectors/clp/search_lib/BaseClpCursor.h" +#include "velox/vector/FlatVector.h" +#include "velox/vector/LazyVector.h" + +namespace facebook::velox::connector::clp::search_lib { + +/// VectorLoader that performs on-demand JSON serialization of IR log events. +/// Converts KeyValuePairLogEvent instances into JSON string representations +/// when loading lazy vectors for __json_string columns. +class ClpIrJsonStringVectorLoader : public VectorLoader { + public: + ClpIrJsonStringVectorLoader( + const std::shared_ptr< + const std::vector>>& + filteredLogEvents) + : filteredLogEvents_(filteredLogEvents) {} + + private: + void loadInternal( + RowSet rows, + ValueHook* hook, + vector_size_t resultSize, + VectorPtr* result) override; + + std::shared_ptr< + const std::vector>> + filteredLogEvents_; +}; + +} // namespace facebook::velox::connector::clp::search_lib diff --git a/velox/connectors/clp/tests/ClpConnectorTest.cpp b/velox/connectors/clp/tests/ClpConnectorTest.cpp index 0018d590b02..4f2963fe97b 100644 --- a/velox/connectors/clp/tests/ClpConnectorTest.cpp +++ b/velox/connectors/clp/tests/ClpConnectorTest.cpp @@ -207,6 +207,82 @@ TEST_F(ClpConnectorTest, test1Pushdown) { test::assertEqualVectors(expected, irOutput); } +TEST_F(ClpConnectorTest, test1JsonString) { + const std::shared_ptr kqlQuery = nullptr; + auto plan = PlanBuilder() + .startTableScan() + .outputType( + ROW({"requestId", "__json_string", "method"}, + {VARCHAR(), VARCHAR(), VARCHAR()})) + .tableHandle(std::make_shared( + kClpConnectorId, "test_1")) + .assignments({ + {"requestId", + std::make_shared( + "requestId", "requestId", VARCHAR())}, + {"__json_string", + std::make_shared( + "__json_string", "__json_string", VARCHAR())}, + {"method", + std::make_shared( + "method", "method", VARCHAR())}, + }) + .endTableScan() + .filter("method = 'GET'") + .planNode(); + + const auto methodVector = makeFlatVector({ + "GET", + "GET", + "GET", + "GET", + "GET", + }); + auto output = getResults( + plan, + {makeClpSplit( + getExampleFilePath("test_1.clps"), + ClpConnectorSplit::SplitType::kArchive, + kqlQuery)}); + auto expected = makeRowVector( + {// requestId + makeFlatVector( + {"req-100", "req-105", "req-107", "req-109", "req-102"}), + // __json_string + makeFlatVector({ + 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 + methodVector}); + test::assertEqualVectors(expected, output); + + auto irOutput = getResults( + plan, + {makeClpSplit( + getExampleFilePath("test_1_ir.clp.zst"), + ClpConnectorSplit::SplitType::kIr, + kqlQuery)}); + auto irExpected = makeRowVector( + {// requestId + makeFlatVector( + {"req-100", "req-102", "req-105", "req-107", "req-109"}), + // __json_string + makeFlatVector({ + R"({"method":"GET","path":"/api/users/1","requestId":"req-100","responseTimeMs":25,"status":200,"timestamp":"2025-04-30T08:45:00Z","userId":"user201"})", + R"({"method":"GET","path":"/public/products","requestId":"req-102","responseTimeMs":18,"status":200,"timestamp":"2025-04-30T08:45:10Z","userId":null})", + R"({"method":"GET","path":"/api/dashboard","requestId":"req-105","responseTimeMs":155,"status":200,"timestamp":"2025-04-30T08:45:25Z","userId":"user204"})", + R"({"method":"GET","path":"/api/users/2/details","requestId":"req-107","responseTimeMs":41,"status":200,"timestamp":"2025-04-30T08:45:35Z","userId":"user202"})", + R"({"method":"GET","path":"/api/products?category=books","requestId":"req-109","responseTimeMs":88,"status":200,"timestamp":"2025-04-30T08:45:45Z","userId":"user203"})", + }), + // method + methodVector}); + test::assertEqualVectors(irExpected, irOutput); +} + TEST_F(ClpConnectorTest, test2NoPushdown) { const std::shared_ptr kqlQuery = nullptr; auto plan = @@ -384,6 +460,59 @@ TEST_F(ClpConnectorTest, test2Hybrid) { test::assertEqualVectors(expected, irOutput); } +TEST_F(ClpConnectorTest, test2JsonString) { + auto kqlQuery = std::make_shared( + "(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( + kClpConnectorId, "test_2")) + .assignments( + {{"timestamp", + std::make_shared( + "timestamp", "timestamp", TIMESTAMP())}, + {"__json_string", + std::make_shared( + "__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(kTestTimestampSeconds, kTestTimestampNanoseconds)}), + // __json_string + makeFlatVector( + {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); + + auto irOutput = getResults( + plan, + {makeClpSplit( + getExampleFilePath("test_2_ir.clp.zst"), + ClpConnectorSplit::SplitType::kIr, + kqlQuery)}); + auto irExpected = makeRowVector( + {// timestamp + makeFlatVector( + {Timestamp(kTestTimestampSeconds, kTestTimestampNanoseconds)}), + // __json_string + makeFlatVector( + {R"({"event":{"details":{"mount":"/var/log","usage":{"percent":92}},"severity":"WARNING","subtype":"disk_usage","tags":["filesystem","monitoring"],"type":"storage"},"timestamp":1746003005})"})}); + + test::assertEqualVectors(irExpected, irOutput); +} + TEST_F(ClpConnectorTest, test3TimestampMarshalling) { const std::shared_ptr kqlQuery = nullptr; auto plan = PlanBuilder(pool_.get())