From fb53c9f42549607d50c37337b0e776e7a1698fe1 Mon Sep 17 00:00:00 2001 From: wraymo Date: Fri, 17 Oct 2025 10:25:40 -0400 Subject: [PATCH 01/11] add clp_get_json_string support --- CMake/resolve_dependency_modules/clp.cmake | 2 +- .../clp/search_lib/archive/CMakeLists.txt | 6 +-- .../search_lib/archive/ClpArchiveCursor.cpp | 44 ++++++++++++---- .../clp/search_lib/archive/ClpArchiveCursor.h | 8 ++- .../ClpArchiveJsonStringVectorLoader.cpp | 50 +++++++++++++++++++ .../ClpArchiveJsonStringVectorLoader.h | 44 ++++++++++++++++ 6 files changed, 137 insertions(+), 17 deletions(-) create mode 100644 velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.cpp create mode 100644 velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h 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/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..060591973ff 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp +++ b/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; @@ -149,7 +151,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 ("__json_string" == column.name) { + jsonStringColumnIndices_.insert(i); + continue; + } std::vector descriptorTokens; std::string descriptorNamespace; if (false == @@ -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,23 @@ 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( + 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_[projectedColumnIndex_].type; + projectedColumnIndex_++; 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..bdec2fdf405 --- /dev/null +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.cpp @@ -0,0 +1,50 @@ +/* +* 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"); + + auto vector = *result; + auto *stringVector = vector->asFlatVector(); + 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 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..d17d53d8cc1 --- /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 + +#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>& 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 From 71b26bef056aaa062d548dd42a1b811aadcb8dff Mon Sep 17 00:00:00 2001 From: wraymo Date: Sat, 18 Oct 2025 10:40:17 -0400 Subject: [PATCH 02/11] add tests --- .../search_lib/archive/ClpArchiveCursor.cpp | 9 +- .../ClpArchiveJsonStringVectorLoader.cpp | 33 ++++--- .../ClpArchiveJsonStringVectorLoader.h | 28 +++--- .../connectors/clp/tests/ClpConnectorTest.cpp | 89 +++++++++++++++++++ 4 files changed, 124 insertions(+), 35 deletions(-) diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp index 060591973ff..813ac18231d 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp @@ -101,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(), @@ -202,7 +202,6 @@ ErrorCode ClpArchiveCursor::loadSplit() { return ErrorCode::InternalError; } projection_->resolve_columns(schemaTree); - archiveReader_->set_projection(projection_); archiveReader_->read_metadata(); @@ -272,8 +271,10 @@ VectorPtr ClpArchiveCursor::createVectorHelper( } VELOX_CHECK_LT( - projectedColumnIndex_, projectedColumns.size(), "Projected column index out of bounds"); - auto *projectedColumn = projectedColumns[projectedColumnIndex_]; + projectedColumnIndex_, + projectedColumns.size(), + "Projected column index out of bounds"); + auto* projectedColumn = projectedColumns[projectedColumnIndex_]; auto projectedType = outputColumns_[projectedColumnIndex_].type; projectedColumnIndex_++; return std::make_shared( diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.cpp b/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.cpp index bdec2fdf405..c1770acaf96 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.cpp +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.cpp @@ -1,18 +1,18 @@ /* -* 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. -*/ + * 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 @@ -25,8 +25,7 @@ namespace facebook::velox::connector::clp::search_lib { ClpArchiveJsonStringVectorLoader::ClpArchiveJsonStringVectorLoader( clp_s::SchemaReader* schemaReader, const std::shared_ptr>& filteredRowIndices) - : schemaReader_(schemaReader), - filteredRowIndices_(filteredRowIndices) { + : schemaReader_(schemaReader), filteredRowIndices_(filteredRowIndices) { schemaReader_->initialize_serializer(); } @@ -38,7 +37,7 @@ void ClpArchiveJsonStringVectorLoader::loadInternal( VELOX_CHECK_NOT_NULL(result, "result vector must not be null"); auto vector = *result; - auto *stringVector = vector->asFlatVector(); + auto* stringVector = vector->asFlatVector(); for (int const vectorIndex : rows) { auto messageIndex = filteredRowIndices_->at(vectorIndex); auto jsonString = schemaReader_->generate_json_string(messageIndex); diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h b/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h index d17d53d8cc1..6127aa8ae5c 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h @@ -1,18 +1,18 @@ /* -* 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. -*/ + * 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 diff --git a/velox/connectors/clp/tests/ClpConnectorTest.cpp b/velox/connectors/clp/tests/ClpConnectorTest.cpp index f31b100aa5d..f54350d32bb 100644 --- a/velox/connectors/clp/tests/ClpConnectorTest.cpp +++ b/velox/connectors/clp/tests/ClpConnectorTest.cpp @@ -207,6 +207,58 @@ 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(); + + 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})", + }), + makeFlatVector({ + "GET", + "GET", + "GET", + "GET", + "GET", + })}); + test::assertEqualVectors(expected, output); +} + TEST_F(ClpConnectorTest, test2NoPushdown) { const std::shared_ptr kqlQuery = nullptr; auto plan = @@ -384,6 +436,43 @@ TEST_F(ClpConnectorTest, test2Hybrid) { test::assertEqualVectors(expected, irOutput); } +TEST_F(ClpConnectorTest, test2JsonSring) { + 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); +} + TEST_F(ClpConnectorTest, test3TimestampMarshalling) { const std::shared_ptr kqlQuery = nullptr; auto plan = PlanBuilder(pool_.get()) From ca77cf325360f4d127f20da153b4afa534b6a963 Mon Sep 17 00:00:00 2001 From: wraymo Date: Sat, 18 Oct 2025 10:41:04 -0400 Subject: [PATCH 03/11] format fix --- .../clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h | 2 ++ 1 file changed, 2 insertions(+) diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h b/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h index 6127aa8ae5c..c606053c3c6 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h @@ -23,6 +23,7 @@ #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 { @@ -41,4 +42,5 @@ class ClpArchiveJsonStringVectorLoader : public VectorLoader { vector_size_t resultSize, VectorPtr* result) override; }; + } // namespace facebook::velox::connector::clp::search_lib From e175675319f02c969e4092cef0a8c7106eaacdcf Mon Sep 17 00:00:00 2001 From: wraymo Date: Sat, 18 Oct 2025 10:42:09 -0400 Subject: [PATCH 04/11] add field annoatation --- velox/connectors/clp/tests/ClpConnectorTest.cpp | 1 + 1 file changed, 1 insertion(+) diff --git a/velox/connectors/clp/tests/ClpConnectorTest.cpp b/velox/connectors/clp/tests/ClpConnectorTest.cpp index f54350d32bb..938a8cf6e17 100644 --- a/velox/connectors/clp/tests/ClpConnectorTest.cpp +++ b/velox/connectors/clp/tests/ClpConnectorTest.cpp @@ -249,6 +249,7 @@ TEST_F(ClpConnectorTest, test1JsonString) { 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({ "GET", "GET", From 3095fa00b28cb0ae793238dc19cb91742cced031 Mon Sep 17 00:00:00 2001 From: wraymo Date: Mon, 20 Oct 2025 10:53:32 -0400 Subject: [PATCH 05/11] address coderabbit comments --- .../clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h | 2 -- velox/connectors/clp/tests/ClpConnectorTest.cpp | 2 +- 2 files changed, 1 insertion(+), 3 deletions(-) diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h b/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h index c606053c3c6..a0df7687a59 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h @@ -16,8 +16,6 @@ #pragma once -#include - #include "clp_s/SchemaReader.hpp" #include "velox/vector/FlatVector.h" #include "velox/vector/LazyVector.h" diff --git a/velox/connectors/clp/tests/ClpConnectorTest.cpp b/velox/connectors/clp/tests/ClpConnectorTest.cpp index b00b76ae5db..415b8283398 100644 --- a/velox/connectors/clp/tests/ClpConnectorTest.cpp +++ b/velox/connectors/clp/tests/ClpConnectorTest.cpp @@ -437,7 +437,7 @@ TEST_F(ClpConnectorTest, test2Hybrid) { test::assertEqualVectors(expected, irOutput); } -TEST_F(ClpConnectorTest, test2JsonSring) { +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 " From ee38e833801b830d2bcedf6cd3831a92c662c60c Mon Sep 17 00:00:00 2001 From: wraymo Date: Tue, 21 Oct 2025 22:33:20 -0400 Subject: [PATCH 06/11] add ir support --- .../clp/search_lib/ir/CMakeLists.txt | 6 +-- .../clp/search_lib/ir/ClpIrCursor.cpp | 35 ++++++++---- .../clp/search_lib/ir/ClpIrCursor.h | 6 ++- .../ir/ClpIrJsonStringVectorLoader.cpp | 50 +++++++++++++++++ .../ir/ClpIrJsonStringVectorLoader.h | 46 ++++++++++++++++ .../connectors/clp/tests/ClpConnectorTest.cpp | 53 ++++++++++++++++--- 6 files changed, 174 insertions(+), 22 deletions(-) create mode 100644 velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.cpp create mode 100644 velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.h 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..acd0f2b71c2 100644 --- a/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp +++ b/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp @@ -19,6 +19,7 @@ #include "ffi/ir_stream/search/QueryHandler.hpp" #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 +27,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 +61,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 +122,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 ("__json_string" == column.name) { + jsonStringColumnIndices_.insert(i); + continue; + } search::ast::literal_type_bitmask_t literalType; switch (column.type) { case ColumnType::Array: @@ -196,17 +202,28 @@ VectorPtr ClpIrCursor::createVectorHelper( auto vector = BaseVector::create(vectorType, vectorSize, pool); vector->setNulls(allocateNulls(vectorSize, pool, bits::kNull)); VELOX_CHECK_LT( - readerIndex_, outputColumns_.size(), "Reader index out of bounds"); - auto projectedColumn = outputColumns_[readerIndex_]; + columnIndex_, outputColumns_.size(), "column index out of bounds"); + const bool isJsonString = jsonStringColumnIndices_.contains(columnIndex_); + ++columnIndex_; + if (isJsonString) { + return std::make_shared( + pool, + vectorType, + vectorSize, + std::make_unique(filteredLogEvents_), + std::move(vector)); + } + + auto projectedColumn = outputColumns_[projectedColumnIndex_]; 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_++; 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..2f5ac039d58 100644 --- a/velox/connectors/clp/search_lib/ir/ClpIrCursor.h +++ b/velox/connectors/clp/search_lib/ir/ClpIrCursor.h @@ -85,11 +85,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..c949cef0f28 --- /dev/null +++ b/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.cpp @@ -0,0 +1,50 @@ +/* + * 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"); + + auto vector = *result; + auto* stringVector = vector->asFlatVector(); + for (int 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 \ No newline at end of file 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..02c5709700d --- /dev/null +++ b/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.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 "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 { + +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 \ No newline at end of file diff --git a/velox/connectors/clp/tests/ClpConnectorTest.cpp b/velox/connectors/clp/tests/ClpConnectorTest.cpp index 415b8283398..4f2963fe97b 100644 --- a/velox/connectors/clp/tests/ClpConnectorTest.cpp +++ b/velox/connectors/clp/tests/ClpConnectorTest.cpp @@ -231,6 +231,13 @@ TEST_F(ClpConnectorTest, test1JsonString) { .filter("method = 'GET'") .planNode(); + const auto methodVector = makeFlatVector({ + "GET", + "GET", + "GET", + "GET", + "GET", + }); auto output = getResults( plan, {makeClpSplit( @@ -250,14 +257,30 @@ TEST_F(ClpConnectorTest, test1JsonString) { R"({"timestamp":"2025-04-30T08:45:10Z","requestId":"req-102","method":"GET","path":"/public/products","responseTimeMs":18,"status":200,"userId":null})", }), // method - makeFlatVector({ - "GET", - "GET", - "GET", - "GET", - "GET", - })}); + 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) { @@ -472,6 +495,22 @@ TEST_F(ClpConnectorTest, test2JsonString) { 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) { From 4b26c7c22bb1a6569b1227472d49b665cf27b07a Mon Sep 17 00:00:00 2001 From: wraymo Date: Tue, 21 Oct 2025 22:54:41 -0400 Subject: [PATCH 07/11] add empty line --- .../clp/search_lib/ir/ClpIrJsonStringVectorLoader.cpp | 2 +- .../connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.h | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.cpp b/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.cpp index c949cef0f28..28fa527c939 100644 --- a/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.cpp +++ b/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.cpp @@ -47,4 +47,4 @@ void ClpIrJsonStringVectorLoader::loadInternal( } } -} // namespace facebook::velox::connector::clp::search_lib \ No newline at end of file +} // 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 index 02c5709700d..b16dc40e910 100644 --- a/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.h +++ b/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.h @@ -43,4 +43,4 @@ class ClpIrJsonStringVectorLoader : public VectorLoader { filteredLogEvents_; }; -} // namespace facebook::velox::connector::clp::search_lib \ No newline at end of file +} // namespace facebook::velox::connector::clp::search_lib From 2c2760531f75286504d5ed2679de0da9fe3cb667 Mon Sep 17 00:00:00 2001 From: wraymo Date: Wed, 22 Oct 2025 15:51:09 -0400 Subject: [PATCH 08/11] address comments --- .../archive/ClpArchiveJsonStringVectorLoader.cpp | 4 +++- .../search_lib/archive/ClpArchiveJsonStringVectorLoader.h | 2 +- velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp | 7 +++++-- velox/connectors/clp/search_lib/ir/ClpIrCursor.h | 2 ++ .../clp/search_lib/ir/ClpIrJsonStringVectorLoader.cpp | 4 +++- .../clp/search_lib/ir/ClpIrJsonStringVectorLoader.h | 3 +++ 6 files changed, 17 insertions(+), 5 deletions(-) diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.cpp b/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.cpp index c1770acaf96..5ba9dbb467d 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.cpp +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.cpp @@ -35,10 +35,12 @@ void ClpArchiveJsonStringVectorLoader::loadInternal( 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 (int const vectorIndex : rows) { + for (vector_size_t const vectorIndex : rows) { auto messageIndex = filteredRowIndices_->at(vectorIndex); auto jsonString = schemaReader_->generate_json_string(messageIndex); stringVector->set(vectorIndex, StringView(jsonString)); diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h b/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h index a0df7687a59..21aa6a87377 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveJsonStringVectorLoader.h @@ -23,7 +23,7 @@ namespace facebook::velox::connector::clp::search_lib { /// A custom Velox VectorLoader that populates Velox vectors with serialized -/// JSON string from CLP column readers. +/// JSON string from CLP column readers for the archive format. class ClpArchiveJsonStringVectorLoader : public VectorLoader { public: ClpArchiveJsonStringVectorLoader( diff --git a/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp b/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp index acd0f2b71c2..9498fc9ddda 100644 --- a/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp +++ b/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp @@ -201,8 +201,7 @@ VectorPtr ClpIrCursor::createVectorHelper( } auto vector = BaseVector::create(vectorType, vectorSize, pool); vector->setNulls(allocateNulls(vectorSize, pool, bits::kNull)); - VELOX_CHECK_LT( - columnIndex_, outputColumns_.size(), "column index out of bounds"); + const bool isJsonString = jsonStringColumnIndices_.contains(columnIndex_); ++columnIndex_; if (isJsonString) { @@ -214,6 +213,10 @@ VectorPtr ClpIrCursor::createVectorHelper( std::move(vector)); } + VELOX_CHECK_LT( + projectedColumnIndex_, + outputColumns_.size(), + "Projected column index out of bounds"); auto projectedColumn = outputColumns_[projectedColumnIndex_]; auto projectedColumnType = projectedColumn.type; auto it = projectedColumnIdxNodeIdsMap_.find(projectedColumnIndex_); diff --git a/velox/connectors/clp/search_lib/ir/ClpIrCursor.h b/velox/connectors/clp/search_lib/ir/ClpIrCursor.h index 2f5ac039d58..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" diff --git a/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.cpp b/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.cpp index 28fa527c939..650b4806bb7 100644 --- a/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.cpp +++ b/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.cpp @@ -27,10 +27,12 @@ void ClpIrJsonStringVectorLoader::loadInternal( 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 (int const vectorIndex : rows) { + for (vector_size_t const vectorIndex : rows) { const auto& logEvent = filteredLogEvents_->at(vectorIndex); auto serializedResult = logEvent->serialize_to_json(); if (serializedResult.has_error()) { diff --git a/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.h b/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.h index b16dc40e910..eb3ff7b4a27 100644 --- a/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.h +++ b/velox/connectors/clp/search_lib/ir/ClpIrJsonStringVectorLoader.h @@ -23,6 +23,9 @@ 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( From f13eb81d8ac1f53c57ccb305a59dba1ead1b96c6 Mon Sep 17 00:00:00 2001 From: wraymo Date: Thu, 23 Oct 2025 09:58:16 -0400 Subject: [PATCH 09/11] add constant for json string column name --- velox/connectors/clp/ClpColumnHandle.h | 2 ++ velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp | 3 ++- velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp | 3 ++- 3 files changed, 6 insertions(+), 2 deletions(-) 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/ClpArchiveCursor.cpp b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp index 813ac18231d..6ea5ae4e1d7 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp @@ -20,6 +20,7 @@ #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" @@ -153,7 +154,7 @@ ErrorCode ClpArchiveCursor::loadSplit() { try { for (size_t i = 0; i < outputColumns_.size(); i++) { auto const& column = outputColumns_[i]; - if ("__json_string" == column.name) { + if (ClpColumnHandle::jsonStringColumnName_ == column.name) { jsonStringColumnIndices_.insert(i); continue; } diff --git a/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp b/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp index 9498fc9ddda..38b18545038 100644 --- a/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp +++ b/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/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" @@ -127,7 +128,7 @@ ClpIrCursor::splitFieldsToNamesAndTypes() { std::pair>{}; for (size_t i{0}; i < outputColumns_.size(); ++i) { auto column = outputColumns_[i]; - if ("__json_string" == column.name) { + if (ClpColumnHandle::jsonStringColumnName_ == column.name) { jsonStringColumnIndices_.insert(i); continue; } From 9b5617036ff00ca75083981515f7bcec6dd64585 Mon Sep 17 00:00:00 2001 From: wraymo Date: Thu, 23 Oct 2025 16:17:22 -0400 Subject: [PATCH 10/11] fix --- .../clp/search_lib/archive/ClpArchiveCursor.cpp | 8 ++++---- velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp | 8 ++++---- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp index 6ea5ae4e1d7..88c4f0b40c7 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp @@ -259,9 +259,8 @@ 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) { + if (jsonStringColumnIndices_.contains(columnIndex_)) { + ++columnIndex_; return std::make_shared( pool, vectorType, @@ -276,8 +275,9 @@ VectorPtr ClpArchiveCursor::createVectorHelper( projectedColumns.size(), "Projected column index out of bounds"); auto* projectedColumn = projectedColumns[projectedColumnIndex_]; - auto projectedType = outputColumns_[projectedColumnIndex_].type; + auto projectedType = outputColumns_[columnIndex_].type; projectedColumnIndex_++; + columnIndex_++; return std::make_shared( pool, vectorType, diff --git a/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp b/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp index 38b18545038..8e4f2ef91d2 100644 --- a/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp +++ b/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp @@ -203,9 +203,8 @@ VectorPtr ClpIrCursor::createVectorHelper( auto vector = BaseVector::create(vectorType, vectorSize, pool); vector->setNulls(allocateNulls(vectorSize, pool, bits::kNull)); - const bool isJsonString = jsonStringColumnIndices_.contains(columnIndex_); - ++columnIndex_; - if (isJsonString) { + if (jsonStringColumnIndices_.contains(columnIndex_)) { + ++columnIndex_; return std::make_shared( pool, vectorType, @@ -218,7 +217,7 @@ VectorPtr ClpIrCursor::createVectorHelper( projectedColumnIndex_, outputColumns_.size(), "Projected column index out of bounds"); - auto projectedColumn = outputColumns_[projectedColumnIndex_]; + auto projectedColumn = outputColumns_[columnIndex_]; auto projectedColumnType = projectedColumn.type; auto it = projectedColumnIdxNodeIdsMap_.find(projectedColumnIndex_); std::vector<::clp::ffi::SchemaTree::Node::id_t> projectedColumnNodeIds{}; @@ -228,6 +227,7 @@ VectorPtr ClpIrCursor::createVectorHelper( projectedColumnNodeIds = it->second; } projectedColumnIndex_++; + columnIndex_++; return std::make_shared( pool, vectorType, From 791f03edd963112a27387d3cb637b73ae48518e7 Mon Sep 17 00:00:00 2001 From: wraymo Date: Thu, 23 Oct 2025 16:36:22 -0400 Subject: [PATCH 11/11] another fix --- velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp b/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp index 8e4f2ef91d2..f26b1cdd810 100644 --- a/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp +++ b/velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp @@ -214,7 +214,7 @@ VectorPtr ClpIrCursor::createVectorHelper( } VELOX_CHECK_LT( - projectedColumnIndex_, + columnIndex_, outputColumns_.size(), "Projected column index out of bounds"); auto projectedColumn = outputColumns_[columnIndex_];