From 5af7ec318d106f5bdc9a5eebabe9a1184e799501 Mon Sep 17 00:00:00 2001 From: anlowee Date: Tue, 19 Aug 2025 16:13:33 +0000 Subject: [PATCH 01/11] Extract ClpCursor abstract class and move the original ClpCursor implementation to ClpArchiveCursor --- velox/connectors/clp/ClpDataSource.cpp | 23 ++++--- .../connectors/clp/search_lib/CMakeLists.txt | 3 +- .../{ClpCursor.cpp => ClpArchiveCursor.cpp} | 36 +++++------ .../clp/search_lib/ClpArchiveCursor.h | 60 +++++++++++++++++++ velox/connectors/clp/search_lib/ClpCursor.h | 40 ++++++------- 5 files changed, 114 insertions(+), 48 deletions(-) rename velox/connectors/clp/search_lib/{ClpCursor.cpp => ClpArchiveCursor.cpp} (91%) create mode 100644 velox/connectors/clp/search_lib/ClpArchiveCursor.h diff --git a/velox/connectors/clp/ClpDataSource.cpp b/velox/connectors/clp/ClpDataSource.cpp index a5a574eb318..d1600234fed 100644 --- a/velox/connectors/clp/ClpDataSource.cpp +++ b/velox/connectors/clp/ClpDataSource.cpp @@ -20,7 +20,7 @@ #include "velox/connectors/clp/ClpConnectorSplit.h" #include "velox/connectors/clp/ClpDataSource.h" #include "velox/connectors/clp/ClpTableHandle.h" -#include "velox/connectors/clp/search_lib/ClpCursor.h" +#include "velox/connectors/clp/search_lib/ClpArchiveCursor.h" #include "velox/connectors/clp/search_lib/ClpVectorLoader.h" #include "velox/vector/FlatVector.h" @@ -101,12 +101,21 @@ void ClpDataSource::addFieldsRecursively( void ClpDataSource::addSplit(std::shared_ptr split) { auto clpSplit = std::dynamic_pointer_cast(split); - if (storageType_ == ClpConfig::StorageType::kFs) { - cursor_ = std::make_unique( - clp_s::InputSource::Filesystem, clpSplit->path_); - } else if (storageType_ == ClpConfig::StorageType::kS3) { - cursor_ = std::make_unique( - clp_s::InputSource::Network, clpSplit->path_); + clp_s::InputSource inputSource; + if (ClpConfig::StorageType::kFs == storageType_) { + inputSource = clp_s::InputSource::Filesystem; + } else if (ClpConfig::StorageType::kS3 == storageType_) { + inputSource = clp_s::InputSource::Network; + } else { + VELOX_UNREACHABLE(); + } + + if (ClpConnectorSplit::SplitType::kArchive == clpSplit->type_) { + cursor_ = std::make_unique( + inputSource, clpSplit->path_); + } else { + VELOX_UNSUPPORTED( + "Unsupported split type: {}", static_cast(clpSplit->type_)); } auto pushDownQuery = clpSplit->kqlQuery_; diff --git a/velox/connectors/clp/search_lib/CMakeLists.txt b/velox/connectors/clp/search_lib/CMakeLists.txt index 55e68060fc4..80bf99471c8 100644 --- a/velox/connectors/clp/search_lib/CMakeLists.txt +++ b/velox/connectors/clp/search_lib/CMakeLists.txt @@ -14,7 +14,8 @@ velox_add_library( clp-s-search STATIC - ClpCursor.cpp + ClpArchiveCursor.cpp + ClpArchiveCursor.h ClpCursor.h ClpQueryRunner.cpp ClpQueryRunner.h diff --git a/velox/connectors/clp/search_lib/ClpCursor.cpp b/velox/connectors/clp/search_lib/ClpArchiveCursor.cpp similarity index 91% rename from velox/connectors/clp/search_lib/ClpCursor.cpp rename to velox/connectors/clp/search_lib/ClpArchiveCursor.cpp index fa6d67bf22b..9b23c2244d3 100644 --- a/velox/connectors/clp/search_lib/ClpCursor.cpp +++ b/velox/connectors/clp/search_lib/ClpArchiveCursor.cpp @@ -16,6 +16,8 @@ #include +#include "ClpArchiveCursor.h" + #include "clp_s/ArchiveReader.hpp" #include "clp_s/search/EvaluateTimestampIndex.hpp" #include "clp_s/search/ast/ConvertToExists.hpp" @@ -25,27 +27,25 @@ #include "clp_s/search/ast/SearchUtils.hpp" #include "clp_s/search/kql/kql.hpp" -#include "velox/connectors/clp/search_lib/ClpCursor.h" - using namespace clp_s; using namespace clp_s::search; using namespace clp_s::search::ast; namespace facebook::velox::connector::clp::search_lib { -ClpCursor::ClpCursor(InputSource inputSource, std::string archivePath) - : errorCode_(ErrorCode::QueryNotInitialized), - inputSource_(inputSource), - archivePath_(std::move(archivePath)), +ClpArchiveCursor::ClpArchiveCursor( + clp_s::InputSource inputSource, + std::string_view splitPath) + : ClpCursor(inputSource, splitPath), archiveReader_(std::make_shared()) {} -ClpCursor::~ClpCursor() { - if (currentArchiveLoaded_) { +ClpArchiveCursor::~ClpArchiveCursor() { + if (currentSplitLoaded_) { archiveReader_->close(); } } -void ClpCursor::executeQuery( +void ClpArchiveCursor::executeQuery( const std::string& query, const std::vector& outputColumns) { query_ = query; @@ -53,21 +53,21 @@ void ClpCursor::executeQuery( errorCode_ = preprocessQuery(); } -uint64_t ClpCursor::fetchNext( +uint64_t ClpArchiveCursor::fetchNext( uint64_t numRows, const std::shared_ptr>& filteredRowIndices) { if (ErrorCode::Success != errorCode_) { return 0; } - if (false == currentArchiveLoaded_) { - errorCode_ = loadArchive(); + if (false == currentSplitLoaded_) { + errorCode_ = loadSplit(); if (ErrorCode::Success != errorCode_) { return 0; } archiveReader_->open_packed_streams(); - currentArchiveLoaded_ = true; + currentSplitLoaded_ = true; queryRunner_ = std::make_shared( schemaMatch_, expr_, archiveReader_, false, projection_); queryRunner_->global_init(); @@ -104,8 +104,8 @@ uint64_t ClpCursor::fetchNext( return 0; } -const std::vector& ClpCursor::getProjectedColumns() - const { +const std::vector& +ClpArchiveCursor::getProjectedColumns() const { if (queryRunner_) { return queryRunner_->getProjectedColumns(); } @@ -113,7 +113,7 @@ const std::vector& ClpCursor::getProjectedColumns() return kEmpty; } -ErrorCode ClpCursor::preprocessQuery() { +ErrorCode ClpArchiveCursor::preprocessQuery() { auto queryStream = std::istringstream(query_); expr_ = kql::parse_kql_expression(queryStream); if (nullptr == expr_) { @@ -150,14 +150,14 @@ ErrorCode ClpCursor::preprocessQuery() { return ErrorCode::Success; } -ErrorCode ClpCursor::loadArchive() { +ErrorCode ClpArchiveCursor::loadSplit() { auto networkAuthOption = inputSource_ == InputSource::Filesystem ? NetworkAuthOption{.method = AuthMethod::None} : NetworkAuthOption{.method = AuthMethod::S3PresignedUrlV4}; try { archiveReader_->open( - get_path_object_for_raw_path(archivePath_), networkAuthOption); + get_path_object_for_raw_path(splitPath_), networkAuthOption); } catch (std::exception& e) { VLOG(2) << "Failed to open archive file: " << e.what(); return ErrorCode::InternalError; diff --git a/velox/connectors/clp/search_lib/ClpArchiveCursor.h b/velox/connectors/clp/search_lib/ClpArchiveCursor.h new file mode 100644 index 00000000000..9466bfda0fe --- /dev/null +++ b/velox/connectors/clp/search_lib/ClpArchiveCursor.h @@ -0,0 +1,60 @@ +/* + * 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 "velox/connectors/clp/search_lib/ClpCursor.h" + +namespace facebook::velox::connector::clp::search_lib { + +class ClpArchiveCursor final : public ClpCursor { + public: + explicit ClpArchiveCursor( + clp_s::InputSource inputSource, + std::string_view splitPath); + ~ClpArchiveCursor() override; + + void executeQuery( + const std::string& query, + const std::vector& outputColumns) override; + + uint64_t fetchNext( + uint64_t numRows, + const std::shared_ptr>& filteredRowIndices) + override; + + const std::vector& getProjectedColumns() + const override; + + protected: + ErrorCode preprocessQuery() override; + + ErrorCode loadSplit() override; + + private: + std::vector matchedSchemas_; + size_t currentSchemaIndex_{0}; + int32_t currentSchemaId_{-1}; + bool currentSchemaTableLoaded_{false}; + + std::shared_ptr schemaMatch_; + std::shared_ptr queryRunner_; + std::shared_ptr projection_; + + std::shared_ptr archiveReader_; +}; + +} // namespace facebook::velox::connector::clp::search_lib diff --git a/velox/connectors/clp/search_lib/ClpCursor.h b/velox/connectors/clp/search_lib/ClpCursor.h index f6303a2ef94..55d547706a7 100644 --- a/velox/connectors/clp/search_lib/ClpCursor.h +++ b/velox/connectors/clp/search_lib/ClpCursor.h @@ -71,8 +71,11 @@ struct Field { /// while supporting projection and batch-oriented retrieval of filtered rows. class ClpCursor { public: - explicit ClpCursor(clp_s::InputSource inputSource, std::string archivePath); - ~ClpCursor(); + explicit ClpCursor(clp_s::InputSource inputSource, std::string_view splitPath) + : errorCode_(ErrorCode::QueryNotInitialized), + inputSource_(inputSource), + splitPath_(std::string(splitPath)) {} + virtual ~ClpCursor() = default; /// Executes a query. This function parses, validates, and prepares the given /// query for execution. @@ -80,54 +83,47 @@ class ClpCursor { /// @param query The KQL query to execute. /// @param outputColumns A vector specifying the columns to be included in the /// query result. - void executeQuery( + virtual void executeQuery( const std::string& query, - const std::vector& outputColumns); + const std::vector& outputColumns) = 0; - /// Fetches the next set of rows from the cursor. If the archive and schema + /// Fetches the next set of rows from the cursor. If the split and schema /// are not yet loaded, this function will perform the necessary loading. /// /// @param numRows The maximum number of rows to fetch. /// @param filteredRowIndices A vector of row indices that match the filter. /// @return The number of rows scanned. - uint64_t fetchNext( + virtual uint64_t fetchNext( uint64_t numRows, - const std::shared_ptr>& filteredRowIndices); + const std::shared_ptr>& filteredRowIndices) = 0; /// Retrieves the projected columns. /// /// @return A vector of BaseColumnReader pointers representing the projected /// columns. - const std::vector& getProjectedColumns() const; + virtual const std::vector& getProjectedColumns() + const = 0; - private: + protected: /// Preprocesses the query, performing parsing, validation, and optimization. /// /// @return The error code. - ErrorCode preprocessQuery(); + virtual ErrorCode preprocessQuery() = 0; - /// Loads the archive at the current index. /// /// @return The error code. - ErrorCode loadArchive(); + virtual ErrorCode loadSplit() = 0; ErrorCode errorCode_; clp_s::InputSource inputSource_{clp_s::InputSource::Filesystem}; - std::string archivePath_; + std::string splitPath_; std::string query_; std::vector outputColumns_; - std::vector matchedSchemas_; - size_t currentSchemaIndex_{0}; - int32_t currentSchemaId_{-1}; - bool currentSchemaTableLoaded_{false}; - bool currentArchiveLoaded_{false}; + + bool currentSplitLoaded_{false}; std::shared_ptr expr_; - std::shared_ptr schemaMatch_; - std::shared_ptr queryRunner_; - std::shared_ptr projection_; - std::shared_ptr archiveReader_; }; } // namespace facebook::velox::connector::clp::search_lib From cc82dd7733998185ec266951a18d91765b6243f1 Mon Sep 17 00:00:00 2001 From: anlowee Date: Tue, 19 Aug 2025 19:02:54 +0000 Subject: [PATCH 02/11] Rename the extracrted ClpCursor to BaseClpCursor, and add the skeleton of ClpIrCursor --- velox/connectors/clp/ClpDataSource.cpp | 5 +- velox/connectors/clp/ClpDataSource.h | 4 +- .../clp/search_lib/BaseClpCursor.cpp | 78 +++++++++++++++++++ .../{ClpCursor.h => BaseClpCursor.h} | 33 +++++--- .../connectors/clp/search_lib/CMakeLists.txt | 5 +- .../clp/search_lib/ClpArchiveCursor.cpp | 51 +----------- .../clp/search_lib/ClpArchiveCursor.h | 10 +-- .../connectors/clp/search_lib/ClpIrCursor.cpp | 21 +++++ velox/connectors/clp/search_lib/ClpIrCursor.h | 49 ++++++++++++ .../clp/search_lib/ClpVectorLoader.cpp | 6 +- .../clp/search_lib/ClpVectorLoader.h | 7 +- 11 files changed, 193 insertions(+), 76 deletions(-) create mode 100644 velox/connectors/clp/search_lib/BaseClpCursor.cpp rename velox/connectors/clp/search_lib/{ClpCursor.h => BaseClpCursor.h} (84%) create mode 100644 velox/connectors/clp/search_lib/ClpIrCursor.cpp create mode 100644 velox/connectors/clp/search_lib/ClpIrCursor.h diff --git a/velox/connectors/clp/ClpDataSource.cpp b/velox/connectors/clp/ClpDataSource.cpp index 962ea363ed0..6c702698fd0 100644 --- a/velox/connectors/clp/ClpDataSource.cpp +++ b/velox/connectors/clp/ClpDataSource.cpp @@ -163,7 +163,10 @@ VectorPtr ClpDataSource::createVector( vectorType, vectorSize, std::make_unique( - projectedColumn, projectedType, filteredRows), + projectedColumn, + projectedType, + filteredRows, + cursor_->getSplitType()), std::move(vector)); } diff --git a/velox/connectors/clp/ClpDataSource.h b/velox/connectors/clp/ClpDataSource.h index 70ec4564354..32611525fe2 100644 --- a/velox/connectors/clp/ClpDataSource.h +++ b/velox/connectors/clp/ClpDataSource.h @@ -20,7 +20,7 @@ #include "velox/connectors/Connector.h" #include "velox/connectors/clp/ClpConfig.h" -#include "velox/connectors/clp/search_lib/ClpCursor.h" +#include "velox/connectors/clp/search_lib/BaseClpCursor.h" namespace clp_s { class BaseColumnReader; @@ -102,7 +102,7 @@ class ClpDataSource : public DataSource { std::vector fields_; - std::unique_ptr cursor_; + std::unique_ptr cursor_; std::shared_ptr s3AuthProvider_; }; diff --git a/velox/connectors/clp/search_lib/BaseClpCursor.cpp b/velox/connectors/clp/search_lib/BaseClpCursor.cpp new file mode 100644 index 00000000000..eafcccbf7a7 --- /dev/null +++ b/velox/connectors/clp/search_lib/BaseClpCursor.cpp @@ -0,0 +1,78 @@ +/* + * 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/BaseClpCursor.h" + +#include "clp_s/search/EvaluateTimestampIndex.hpp" +#include "clp_s/search/ast/ConvertToExists.hpp" +#include "clp_s/search/ast/EmptyExpr.hpp" +#include "clp_s/search/ast/NarrowTypes.hpp" +#include "clp_s/search/ast/OrOfAndForm.hpp" +#include "clp_s/search/ast/SearchUtils.hpp" +#include "clp_s/search/kql/kql.hpp" + +using namespace clp_s; +using namespace clp_s::search; +using namespace clp_s::search::ast; + +namespace facebook::velox::connector::clp::search_lib { + +void BaseClpCursor::executeQuery( + const std::string& query, + const std::vector& outputColumns) { + query_ = query; + outputColumns_ = outputColumns; + errorCode_ = preprocessQuery(); +} + +ErrorCode BaseClpCursor::preprocessQuery() { + auto queryStream = std::istringstream(query_); + expr_ = kql::parse_kql_expression(queryStream); + if (nullptr == expr_) { + VLOG(2) << "Failed to parse query '" << query_ << "'"; + return ErrorCode::InvalidQuerySyntax; + } + + if (std::dynamic_pointer_cast(expr_)) { + VLOG(2) << "Query '" << query_ << "' is logically false"; + return ErrorCode::LogicalError; + } + + OrOfAndForm standardizePass; + if (expr_ = standardizePass.run(expr_); + std::dynamic_pointer_cast(expr_)) { + VLOG(2) << "Query '" << query_ << "' is logically false"; + return ErrorCode::LogicalError; + } + + NarrowTypes narrowPass; + if (expr_ = narrowPass.run(expr_); + std::dynamic_pointer_cast(expr_)) { + VLOG(2) << "Query '" << query_ << "' is logically false"; + return ErrorCode::LogicalError; + } + + ConvertToExists convertPass; + if (expr_ = convertPass.run(expr_); + std::dynamic_pointer_cast(expr_)) { + VLOG(2) << "Query '" << query_ << "' is logically false"; + return ErrorCode::LogicalError; + } + + return ErrorCode::Success; +} + +} // namespace facebook::velox::connector::clp::search_lib diff --git a/velox/connectors/clp/search_lib/ClpCursor.h b/velox/connectors/clp/search_lib/BaseClpCursor.h similarity index 84% rename from velox/connectors/clp/search_lib/ClpCursor.h rename to velox/connectors/clp/search_lib/BaseClpCursor.h index 55d547706a7..77785140383 100644 --- a/velox/connectors/clp/search_lib/ClpCursor.h +++ b/velox/connectors/clp/search_lib/BaseClpCursor.h @@ -19,6 +19,7 @@ #include #include +#include "connectors/clp/ClpConnectorSplit.h" #include "velox/connectors/clp/search_lib/ClpQueryRunner.h" namespace clp_s { @@ -69,13 +70,16 @@ struct Field { /// schemas and archives, applying filters, and iterating over the results. It /// abstracts away the low-level details of archive access and schema matching /// while supporting projection and batch-oriented retrieval of filtered rows. -class ClpCursor { +class BaseClpCursor { public: - explicit ClpCursor(clp_s::InputSource inputSource, std::string_view splitPath) + explicit BaseClpCursor( + clp_s::InputSource inputSource, + std::string_view splitPath) : errorCode_(ErrorCode::QueryNotInitialized), inputSource_(inputSource), - splitPath_(std::string(splitPath)) {} - virtual ~ClpCursor() = default; + splitPath_(std::string(splitPath)), + splitType_(ClpConnectorSplit::SplitType::kArchive) {} + virtual ~BaseClpCursor() = default; /// Executes a query. This function parses, validates, and prepares the given /// query for execution. @@ -83,9 +87,9 @@ class ClpCursor { /// @param query The KQL query to execute. /// @param outputColumns A vector specifying the columns to be included in the /// query result. - virtual void executeQuery( + void executeQuery( const std::string& query, - const std::vector& outputColumns) = 0; + const std::vector& outputColumns); /// Fetches the next set of rows from the cursor. If the split and schema /// are not yet loaded, this function will perform the necessary loading. @@ -104,12 +108,14 @@ class ClpCursor { virtual const std::vector& getProjectedColumns() const = 0; - protected: - /// Preprocesses the query, performing parsing, validation, and optimization. + /// Get the type of the split that the cursor is processing. /// - /// @return The error code. - virtual ErrorCode preprocessQuery() = 0; + /// @return The split type. + ClpConnectorSplit::SplitType getSplitType() const { + return splitType_; + } + protected: /// /// @return The error code. virtual ErrorCode loadSplit() = 0; @@ -118,12 +124,19 @@ class ClpCursor { clp_s::InputSource inputSource_{clp_s::InputSource::Filesystem}; std::string splitPath_; + ClpConnectorSplit::SplitType splitType_; std::string query_; std::vector outputColumns_; bool currentSplitLoaded_{false}; std::shared_ptr expr_; + + private: + /// Preprocesses the query, performing parsing, validation, and optimization. + /// + /// @return The error code. + ErrorCode preprocessQuery(); }; } // namespace facebook::velox::connector::clp::search_lib diff --git a/velox/connectors/clp/search_lib/CMakeLists.txt b/velox/connectors/clp/search_lib/CMakeLists.txt index c411181bb88..ae0ed3e9ea2 100644 --- a/velox/connectors/clp/search_lib/CMakeLists.txt +++ b/velox/connectors/clp/search_lib/CMakeLists.txt @@ -14,9 +14,12 @@ velox_add_library( clp-s-search STATIC + BaseClpCursor.cpp + BaseClpCursor.h ClpArchiveCursor.cpp ClpArchiveCursor.h - ClpCursor.h + ClpIrCursor.cpp + ClpIrCursor.h ClpPackageS3AuthProvider.cpp ClpPackageS3AuthProvider.h ClpQueryRunner.cpp diff --git a/velox/connectors/clp/search_lib/ClpArchiveCursor.cpp b/velox/connectors/clp/search_lib/ClpArchiveCursor.cpp index 9b23c2244d3..5f7a0e0b045 100644 --- a/velox/connectors/clp/search_lib/ClpArchiveCursor.cpp +++ b/velox/connectors/clp/search_lib/ClpArchiveCursor.cpp @@ -20,12 +20,8 @@ #include "clp_s/ArchiveReader.hpp" #include "clp_s/search/EvaluateTimestampIndex.hpp" -#include "clp_s/search/ast/ConvertToExists.hpp" #include "clp_s/search/ast/EmptyExpr.hpp" -#include "clp_s/search/ast/NarrowTypes.hpp" -#include "clp_s/search/ast/OrOfAndForm.hpp" #include "clp_s/search/ast/SearchUtils.hpp" -#include "clp_s/search/kql/kql.hpp" using namespace clp_s; using namespace clp_s::search; @@ -36,7 +32,7 @@ namespace facebook::velox::connector::clp::search_lib { ClpArchiveCursor::ClpArchiveCursor( clp_s::InputSource inputSource, std::string_view splitPath) - : ClpCursor(inputSource, splitPath), + : BaseClpCursor(inputSource, splitPath), archiveReader_(std::make_shared()) {} ClpArchiveCursor::~ClpArchiveCursor() { @@ -45,14 +41,6 @@ ClpArchiveCursor::~ClpArchiveCursor() { } } -void ClpArchiveCursor::executeQuery( - const std::string& query, - const std::vector& outputColumns) { - query_ = query; - outputColumns_ = outputColumns; - errorCode_ = preprocessQuery(); -} - uint64_t ClpArchiveCursor::fetchNext( uint64_t numRows, const std::shared_ptr>& filteredRowIndices) { @@ -113,43 +101,6 @@ ClpArchiveCursor::getProjectedColumns() const { return kEmpty; } -ErrorCode ClpArchiveCursor::preprocessQuery() { - auto queryStream = std::istringstream(query_); - expr_ = kql::parse_kql_expression(queryStream); - if (nullptr == expr_) { - VLOG(2) << "Failed to parse query '" << query_ << "'"; - return ErrorCode::InvalidQuerySyntax; - } - - if (std::dynamic_pointer_cast(expr_)) { - VLOG(2) << "Query '" << query_ << "' is logically false"; - return ErrorCode::LogicalError; - } - - OrOfAndForm standardizePass; - if (expr_ = standardizePass.run(expr_); - std::dynamic_pointer_cast(expr_)) { - VLOG(2) << "Query '" << query_ << "' is logically false"; - return ErrorCode::LogicalError; - } - - NarrowTypes narrowPass; - if (expr_ = narrowPass.run(expr_); - std::dynamic_pointer_cast(expr_)) { - VLOG(2) << "Query '" << query_ << "' is logically false"; - return ErrorCode::LogicalError; - } - - ConvertToExists convertPass; - if (expr_ = convertPass.run(expr_); - std::dynamic_pointer_cast(expr_)) { - VLOG(2) << "Query '" << query_ << "' is logically false"; - return ErrorCode::LogicalError; - } - - return ErrorCode::Success; -} - ErrorCode ClpArchiveCursor::loadSplit() { auto networkAuthOption = inputSource_ == InputSource::Filesystem ? NetworkAuthOption{.method = AuthMethod::None} diff --git a/velox/connectors/clp/search_lib/ClpArchiveCursor.h b/velox/connectors/clp/search_lib/ClpArchiveCursor.h index 9466bfda0fe..6167a82db7a 100644 --- a/velox/connectors/clp/search_lib/ClpArchiveCursor.h +++ b/velox/connectors/clp/search_lib/ClpArchiveCursor.h @@ -16,21 +16,17 @@ #pragma once -#include "velox/connectors/clp/search_lib/ClpCursor.h" +#include "velox/connectors/clp/search_lib/BaseClpCursor.h" namespace facebook::velox::connector::clp::search_lib { -class ClpArchiveCursor final : public ClpCursor { +class ClpArchiveCursor final : public BaseClpCursor { public: explicit ClpArchiveCursor( clp_s::InputSource inputSource, std::string_view splitPath); ~ClpArchiveCursor() override; - void executeQuery( - const std::string& query, - const std::vector& outputColumns) override; - uint64_t fetchNext( uint64_t numRows, const std::shared_ptr>& filteredRowIndices) @@ -40,8 +36,6 @@ class ClpArchiveCursor final : public ClpCursor { const override; protected: - ErrorCode preprocessQuery() override; - ErrorCode loadSplit() override; private: diff --git a/velox/connectors/clp/search_lib/ClpIrCursor.cpp b/velox/connectors/clp/search_lib/ClpIrCursor.cpp new file mode 100644 index 00000000000..cf1a6ecec12 --- /dev/null +++ b/velox/connectors/clp/search_lib/ClpIrCursor.cpp @@ -0,0 +1,21 @@ +/* + * 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 "ClpIrCursor.h" + +namespace facebook::velox::connector::clp::search_lib { + +} // namespace facebook::velox::connector::clp::search_lib diff --git a/velox/connectors/clp/search_lib/ClpIrCursor.h b/velox/connectors/clp/search_lib/ClpIrCursor.h new file mode 100644 index 00000000000..82d2e3daf8e --- /dev/null +++ b/velox/connectors/clp/search_lib/ClpIrCursor.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 "streaming_compression/Decompressor.hpp" +#include "streaming_compression/zstd/Decompressor.hpp" +#include "velox/connectors/clp/search_lib/BaseClpCursor.h" + +namespace facebook::velox::connector::clp::search_lib { + +class ClpIrCursor final : public BaseClpCursor { + public: + explicit ClpIrCursor( + clp_s::InputSource inputSource, + std::string_view splitPath); + ~ClpIrCursor() override; + + uint64_t fetchNext( + uint64_t numRows, + const std::shared_ptr>& filteredRowIndices) + override; + + const std::vector& getProjectedColumns() + const override; + + protected: + ErrorCode loadSplit() override; + + private: + std::shared_ptr<::clp::ReaderInterface> ir_reader_; + std::shared_ptr<::clp::streaming_compression::zstd::Decompressor> + ir_decompressor_; +}; + +} // namespace facebook::velox::connector::clp::search_lib diff --git a/velox/connectors/clp/search_lib/ClpVectorLoader.cpp b/velox/connectors/clp/search_lib/ClpVectorLoader.cpp index 6bd58e8a2a2..65ffb64c250 100644 --- a/velox/connectors/clp/search_lib/ClpVectorLoader.cpp +++ b/velox/connectors/clp/search_lib/ClpVectorLoader.cpp @@ -124,10 +124,12 @@ auto convertToVeloxTimestamp(int64_t timestamp) -> Timestamp { ClpVectorLoader::ClpVectorLoader( clp_s::BaseColumnReader* columnReader, ColumnType nodeType, - std::shared_ptr> filteredRowIndices) + std::shared_ptr> filteredRowIndices, + ClpConnectorSplit::SplitType splitType) : columnReader_(columnReader), nodeType_(nodeType), - filteredRowIndices_(std::move(filteredRowIndices)) {} + filteredRowIndices_(std::move(filteredRowIndices)), + splitType_(splitType) {} template void ClpVectorLoader::populateData(RowSet rows, VectorPtr vector) { diff --git a/velox/connectors/clp/search_lib/ClpVectorLoader.h b/velox/connectors/clp/search_lib/ClpVectorLoader.h index 36af6d7b807..0d81b72ff01 100644 --- a/velox/connectors/clp/search_lib/ClpVectorLoader.h +++ b/velox/connectors/clp/search_lib/ClpVectorLoader.h @@ -18,8 +18,9 @@ #include "clp_s/ColumnReader.hpp" #include "clp_s/SchemaTree.hpp" +#include "connectors/clp/ClpConnectorSplit.h" -#include "velox/connectors/clp/search_lib/ClpCursor.h" +#include "velox/connectors/clp/search_lib/BaseClpCursor.h" #include "velox/type/Timestamp.h" #include "velox/vector/FlatVector.h" #include "velox/vector/LazyVector.h" @@ -38,7 +39,8 @@ class ClpVectorLoader : public VectorLoader { ClpVectorLoader( clp_s::BaseColumnReader* columnReader, ColumnType nodeType, - std::shared_ptr> filteredRowIndices); + std::shared_ptr> filteredRowIndices, + ClpConnectorSplit::SplitType splitType); private: void loadInternal( @@ -58,6 +60,7 @@ class ClpVectorLoader : public VectorLoader { clp_s::BaseColumnReader* columnReader_; ColumnType nodeType_; std::shared_ptr> filteredRowIndices_; + ClpConnectorSplit::SplitType splitType_; inline static thread_local std::unique_ptr arrayParser_ = std::make_unique(); From d45baf0489e36b38359e9559d6d555f36fec5657 Mon Sep 17 00:00:00 2001 From: anlowee Date: Wed, 20 Aug 2025 16:20:23 +0000 Subject: [PATCH 03/11] Remove the IR related code and add a subpackage to store archive search related code --- velox/connectors/clp/ClpDataSource.cpp | 2 +- .../connectors/clp/search_lib/BaseClpCursor.h | 2 +- .../connectors/clp/search_lib/CMakeLists.txt | 10 ++-- .../connectors/clp/search_lib/ClpIrCursor.cpp | 21 -------- velox/connectors/clp/search_lib/ClpIrCursor.h | 49 ------------------- .../clp/search_lib/archive/CMakeLists.txt | 19 +++++++ .../{ => archive}/ClpArchiveCursor.cpp | 0 .../{ => archive}/ClpArchiveCursor.h | 0 .../{ => archive}/ClpQueryRunner.cpp | 2 +- .../search_lib/{ => archive}/ClpQueryRunner.h | 0 10 files changed, 26 insertions(+), 79 deletions(-) delete mode 100644 velox/connectors/clp/search_lib/ClpIrCursor.cpp delete mode 100644 velox/connectors/clp/search_lib/ClpIrCursor.h create mode 100644 velox/connectors/clp/search_lib/archive/CMakeLists.txt rename velox/connectors/clp/search_lib/{ => archive}/ClpArchiveCursor.cpp (100%) rename velox/connectors/clp/search_lib/{ => archive}/ClpArchiveCursor.h (100%) rename velox/connectors/clp/search_lib/{ => archive}/ClpQueryRunner.cpp (97%) rename velox/connectors/clp/search_lib/{ => archive}/ClpQueryRunner.h (100%) diff --git a/velox/connectors/clp/ClpDataSource.cpp b/velox/connectors/clp/ClpDataSource.cpp index 6c702698fd0..97cf0557ab9 100644 --- a/velox/connectors/clp/ClpDataSource.cpp +++ b/velox/connectors/clp/ClpDataSource.cpp @@ -17,11 +17,11 @@ #include #include "search_lib/ClpS3AuthProviderBase.h" +#include "search_lib/archive/ClpArchiveCursor.h" #include "velox/connectors/clp/ClpColumnHandle.h" #include "velox/connectors/clp/ClpConnectorSplit.h" #include "velox/connectors/clp/ClpDataSource.h" #include "velox/connectors/clp/ClpTableHandle.h" -#include "velox/connectors/clp/search_lib/ClpArchiveCursor.h" #include "velox/connectors/clp/search_lib/ClpVectorLoader.h" #include "velox/vector/FlatVector.h" diff --git a/velox/connectors/clp/search_lib/BaseClpCursor.h b/velox/connectors/clp/search_lib/BaseClpCursor.h index 77785140383..d481708ed77 100644 --- a/velox/connectors/clp/search_lib/BaseClpCursor.h +++ b/velox/connectors/clp/search_lib/BaseClpCursor.h @@ -19,8 +19,8 @@ #include #include +#include "archive/ClpQueryRunner.h" #include "connectors/clp/ClpConnectorSplit.h" -#include "velox/connectors/clp/search_lib/ClpQueryRunner.h" namespace clp_s { enum class InputSource : uint8_t; diff --git a/velox/connectors/clp/search_lib/CMakeLists.txt b/velox/connectors/clp/search_lib/CMakeLists.txt index ae0ed3e9ea2..823e135ff4c 100644 --- a/velox/connectors/clp/search_lib/CMakeLists.txt +++ b/velox/connectors/clp/search_lib/CMakeLists.txt @@ -16,19 +16,17 @@ velox_add_library( STATIC BaseClpCursor.cpp BaseClpCursor.h - ClpArchiveCursor.cpp - ClpArchiveCursor.h - ClpIrCursor.cpp - ClpIrCursor.h ClpPackageS3AuthProvider.cpp ClpPackageS3AuthProvider.h - ClpQueryRunner.cpp - ClpQueryRunner.h ClpS3AuthProviderBase.cpp ClpS3AuthProviderBase.h ClpVectorLoader.cpp ClpVectorLoader.h) +target_include_directories(clp-s-search PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}) + +add_subdirectory(archive) + velox_link_libraries( clp-s-search PUBLIC clp_s::archive_reader diff --git a/velox/connectors/clp/search_lib/ClpIrCursor.cpp b/velox/connectors/clp/search_lib/ClpIrCursor.cpp deleted file mode 100644 index cf1a6ecec12..00000000000 --- a/velox/connectors/clp/search_lib/ClpIrCursor.cpp +++ /dev/null @@ -1,21 +0,0 @@ -/* - * 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 "ClpIrCursor.h" - -namespace facebook::velox::connector::clp::search_lib { - -} // namespace facebook::velox::connector::clp::search_lib diff --git a/velox/connectors/clp/search_lib/ClpIrCursor.h b/velox/connectors/clp/search_lib/ClpIrCursor.h deleted file mode 100644 index 82d2e3daf8e..00000000000 --- a/velox/connectors/clp/search_lib/ClpIrCursor.h +++ /dev/null @@ -1,49 +0,0 @@ -/* - * 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 "streaming_compression/Decompressor.hpp" -#include "streaming_compression/zstd/Decompressor.hpp" -#include "velox/connectors/clp/search_lib/BaseClpCursor.h" - -namespace facebook::velox::connector::clp::search_lib { - -class ClpIrCursor final : public BaseClpCursor { - public: - explicit ClpIrCursor( - clp_s::InputSource inputSource, - std::string_view splitPath); - ~ClpIrCursor() override; - - uint64_t fetchNext( - uint64_t numRows, - const std::shared_ptr>& filteredRowIndices) - override; - - const std::vector& getProjectedColumns() - const override; - - protected: - ErrorCode loadSplit() override; - - private: - std::shared_ptr<::clp::ReaderInterface> ir_reader_; - std::shared_ptr<::clp::streaming_compression::zstd::Decompressor> - ir_decompressor_; -}; - -} // namespace facebook::velox::connector::clp::search_lib diff --git a/velox/connectors/clp/search_lib/archive/CMakeLists.txt b/velox/connectors/clp/search_lib/archive/CMakeLists.txt new file mode 100644 index 00000000000..3777be57d53 --- /dev/null +++ b/velox/connectors/clp/search_lib/archive/CMakeLists.txt @@ -0,0 +1,19 @@ +# 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. +target_sources( + clp-s-search + PRIVATE ClpArchiveCursor.cpp ClpArchiveCursor.h ClpQueryRunner.cpp + ClpQueryRunner.h) + +target_include_directories(clp-s-search PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}) diff --git a/velox/connectors/clp/search_lib/ClpArchiveCursor.cpp b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp similarity index 100% rename from velox/connectors/clp/search_lib/ClpArchiveCursor.cpp rename to velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp diff --git a/velox/connectors/clp/search_lib/ClpArchiveCursor.h b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h similarity index 100% rename from velox/connectors/clp/search_lib/ClpArchiveCursor.h rename to velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h diff --git a/velox/connectors/clp/search_lib/ClpQueryRunner.cpp b/velox/connectors/clp/search_lib/archive/ClpQueryRunner.cpp similarity index 97% rename from velox/connectors/clp/search_lib/ClpQueryRunner.cpp rename to velox/connectors/clp/search_lib/archive/ClpQueryRunner.cpp index a1c9032ebd5..dcb6fdc4bb5 100644 --- a/velox/connectors/clp/search_lib/ClpQueryRunner.cpp +++ b/velox/connectors/clp/search_lib/archive/ClpQueryRunner.cpp @@ -14,7 +14,7 @@ * limitations under the License. */ -#include "velox/connectors/clp/search_lib/ClpQueryRunner.h" +#include "ClpQueryRunner.h" using namespace clp_s; using namespace clp_s::search; diff --git a/velox/connectors/clp/search_lib/ClpQueryRunner.h b/velox/connectors/clp/search_lib/archive/ClpQueryRunner.h similarity index 100% rename from velox/connectors/clp/search_lib/ClpQueryRunner.h rename to velox/connectors/clp/search_lib/archive/ClpQueryRunner.h From 7f40b7e041a136ab0e76cf75cd888f384a070961 Mon Sep 17 00:00:00 2001 From: anlowee Date: Wed, 20 Aug 2025 17:40:38 +0000 Subject: [PATCH 04/11] Address coderabbitai comments --- velox/connectors/clp/ClpDataSource.cpp | 4 ++-- velox/connectors/clp/search_lib/BaseClpCursor.cpp | 4 ++-- velox/connectors/clp/search_lib/BaseClpCursor.h | 13 ++++--------- .../connectors/clp/search_lib/ClpVectorLoader.cpp | 1 + velox/connectors/clp/search_lib/ClpVectorLoader.h | 7 +++++-- .../clp/search_lib/archive/ClpArchiveCursor.cpp | 15 ++++++++------- .../clp/search_lib/archive/ClpArchiveCursor.h | 11 +++++++++++ 7 files changed, 33 insertions(+), 22 deletions(-) diff --git a/velox/connectors/clp/ClpDataSource.cpp b/velox/connectors/clp/ClpDataSource.cpp index 97cf0557ab9..03059f584b6 100644 --- a/velox/connectors/clp/ClpDataSource.cpp +++ b/velox/connectors/clp/ClpDataSource.cpp @@ -16,13 +16,13 @@ #include -#include "search_lib/ClpS3AuthProviderBase.h" -#include "search_lib/archive/ClpArchiveCursor.h" #include "velox/connectors/clp/ClpColumnHandle.h" #include "velox/connectors/clp/ClpConnectorSplit.h" #include "velox/connectors/clp/ClpDataSource.h" #include "velox/connectors/clp/ClpTableHandle.h" +#include "velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h" #include "velox/connectors/clp/search_lib/ClpVectorLoader.h" +#include "velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h" #include "velox/vector/FlatVector.h" namespace facebook::velox::connector::clp { diff --git a/velox/connectors/clp/search_lib/BaseClpCursor.cpp b/velox/connectors/clp/search_lib/BaseClpCursor.cpp index eafcccbf7a7..b0c64bf464d 100644 --- a/velox/connectors/clp/search_lib/BaseClpCursor.cpp +++ b/velox/connectors/clp/search_lib/BaseClpCursor.cpp @@ -14,14 +14,14 @@ * limitations under the License. */ +#include + #include "velox/connectors/clp/search_lib/BaseClpCursor.h" -#include "clp_s/search/EvaluateTimestampIndex.hpp" #include "clp_s/search/ast/ConvertToExists.hpp" #include "clp_s/search/ast/EmptyExpr.hpp" #include "clp_s/search/ast/NarrowTypes.hpp" #include "clp_s/search/ast/OrOfAndForm.hpp" -#include "clp_s/search/ast/SearchUtils.hpp" #include "clp_s/search/kql/kql.hpp" using namespace clp_s; diff --git a/velox/connectors/clp/search_lib/BaseClpCursor.h b/velox/connectors/clp/search_lib/BaseClpCursor.h index d481708ed77..aa4152dfbaf 100644 --- a/velox/connectors/clp/search_lib/BaseClpCursor.h +++ b/velox/connectors/clp/search_lib/BaseClpCursor.h @@ -16,23 +16,18 @@ #pragma once +#include #include +#include #include -#include "archive/ClpQueryRunner.h" -#include "connectors/clp/ClpConnectorSplit.h" +#include "clp_s/InputConfig.hpp" +#include "velox/connectors/clp/ClpConnectorSplit.h" namespace clp_s { -enum class InputSource : uint8_t; -class ArchiveReader; class BaseColumnReader; } // namespace clp_s -namespace clp_s::search { -class Projection; -class SchemaMatch; -} // namespace clp_s::search - namespace clp_s::search::ast { class Expression; } // namespace clp_s::search::ast diff --git a/velox/connectors/clp/search_lib/ClpVectorLoader.cpp b/velox/connectors/clp/search_lib/ClpVectorLoader.cpp index 65ffb64c250..1471b66a5e5 100644 --- a/velox/connectors/clp/search_lib/ClpVectorLoader.cpp +++ b/velox/connectors/clp/search_lib/ClpVectorLoader.cpp @@ -21,6 +21,7 @@ #include "clp_s/ColumnReader.hpp" #include "clp_s/SchemaTree.hpp" +#include "velox/connectors/clp/search_lib/BaseClpCursor.h" #include "velox/connectors/clp/search_lib/ClpVectorLoader.h" #include "velox/type/Timestamp.h" #include "velox/vector/ComplexVector.h" diff --git a/velox/connectors/clp/search_lib/ClpVectorLoader.h b/velox/connectors/clp/search_lib/ClpVectorLoader.h index 0d81b72ff01..69a7f259f24 100644 --- a/velox/connectors/clp/search_lib/ClpVectorLoader.h +++ b/velox/connectors/clp/search_lib/ClpVectorLoader.h @@ -16,11 +16,12 @@ #pragma once +#include + #include "clp_s/ColumnReader.hpp" #include "clp_s/SchemaTree.hpp" -#include "connectors/clp/ClpConnectorSplit.h" +#include "velox/connectors/clp/ClpConnectorSplit.h" -#include "velox/connectors/clp/search_lib/BaseClpCursor.h" #include "velox/type/Timestamp.h" #include "velox/vector/FlatVector.h" #include "velox/vector/LazyVector.h" @@ -31,6 +32,8 @@ class BaseColumnReader; namespace facebook::velox::connector::clp::search_lib { +enum class ColumnType; + /// A custom Velox VectorLoader that populates Velox vectors from a CLP-based /// column reader. It supports various column types including integers, floats, /// booleans, strings, and arrays of strings. diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp index 5f7a0e0b045..55c5e995df2 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp @@ -16,12 +16,13 @@ #include -#include "ClpArchiveCursor.h" +#include "velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h" #include "clp_s/ArchiveReader.hpp" #include "clp_s/search/EvaluateTimestampIndex.hpp" #include "clp_s/search/ast/EmptyExpr.hpp" #include "clp_s/search/ast/SearchUtils.hpp" +#include "velox/connectors/clp/search_lib/archive/ClpQueryRunner.h" using namespace clp_s; using namespace clp_s::search; @@ -53,12 +54,6 @@ uint64_t ClpArchiveCursor::fetchNext( if (ErrorCode::Success != errorCode_) { return 0; } - - archiveReader_->open_packed_streams(); - currentSplitLoaded_ = true; - queryRunner_ = std::make_shared( - schemaMatch_, expr_, archiveReader_, false, projection_); - queryRunner_->global_init(); } while (currentSchemaIndex_ < matchedSchemas_.size()) { @@ -207,6 +202,12 @@ ErrorCode ClpArchiveCursor::loadSplit() { currentSchemaIndex_ = 0; currentSchemaTableLoaded_ = false; + + archiveReader_->open_packed_streams(); + currentSplitLoaded_ = true; + queryRunner_ = std::make_shared( + schemaMatch_, expr_, archiveReader_, false, projection_); + queryRunner_->global_init(); return ErrorCode::Success; } diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h index 6167a82db7a..65859bc5b1a 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h @@ -18,8 +18,19 @@ #include "velox/connectors/clp/search_lib/BaseClpCursor.h" +namespace clp_s { +class ArchiveReader; +} // namespace clp_s + +namespace clp_s::search { +class Projection; +class SchemaMatch; +} // namespace clp_s::search + namespace facebook::velox::connector::clp::search_lib { +class ClpQueryRunner; + class ClpArchiveCursor final : public BaseClpCursor { public: explicit ClpArchiveCursor( From c0cac3d302cd1ca6cbaad760086e865d1853c545 Mon Sep 17 00:00:00 2001 From: anlowee Date: Wed, 20 Aug 2025 22:39:26 +0000 Subject: [PATCH 05/11] Address comments --- .../connectors/clp/search_lib/BaseClpCursor.h | 10 ++++----- .../connectors/clp/search_lib/CMakeLists.txt | 13 ++---------- .../clp/search_lib/archive/CMakeLists.txt | 21 ++++++++++++++----- .../clp/search_lib/archive/ClpArchiveCursor.h | 2 ++ .../clp/search_lib/archive/ClpQueryRunner.cpp | 2 +- 5 files changed, 26 insertions(+), 22 deletions(-) diff --git a/velox/connectors/clp/search_lib/BaseClpCursor.h b/velox/connectors/clp/search_lib/BaseClpCursor.h index aa4152dfbaf..91f44a258f8 100644 --- a/velox/connectors/clp/search_lib/BaseClpCursor.h +++ b/velox/connectors/clp/search_lib/BaseClpCursor.h @@ -61,9 +61,9 @@ struct Field { }; /// A query execution interface that manages the lifecycle of a query on a CLP-S -/// archive, including parsing and validating the query, loading the relevant -/// schemas and archives, applying filters, and iterating over the results. It -/// abstracts away the low-level details of archive access and schema matching +/// split (archive or IR), including parsing and validating the query, loading +/// the relevant splits, applying filters, and iterating over the results. It +/// abstracts away the low-level details of split access /// while supporting projection and batch-oriented retrieval of filtered rows. class BaseClpCursor { public: @@ -86,8 +86,8 @@ class BaseClpCursor { const std::string& query, const std::vector& outputColumns); - /// Fetches the next set of rows from the cursor. If the split and schema - /// are not yet loaded, this function will perform the necessary loading. + /// Fetches the next set of rows from the cursor. If the split is not yet + /// loaded, this function will perform the necessary loading. /// /// @param numRows The maximum number of rows to fetch. /// @param filteredRowIndices A vector of row indices that match the filter. diff --git a/velox/connectors/clp/search_lib/CMakeLists.txt b/velox/connectors/clp/search_lib/CMakeLists.txt index 823e135ff4c..4575eec1284 100644 --- a/velox/connectors/clp/search_lib/CMakeLists.txt +++ b/velox/connectors/clp/search_lib/CMakeLists.txt @@ -23,17 +23,8 @@ velox_add_library( ClpVectorLoader.cpp ClpVectorLoader.h) -target_include_directories(clp-s-search PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}) - add_subdirectory(archive) -velox_link_libraries( - clp-s-search - PUBLIC clp_s::archive_reader - PRIVATE - clp_s::clp_dependencies - clp_s::io - clp_s::search - clp_s::search::kql - velox_vector) +velox_link_libraries(clp-s-search PUBLIC clp-s-archive-search) + target_compile_features(clp-s-search PRIVATE cxx_std_20) diff --git a/velox/connectors/clp/search_lib/archive/CMakeLists.txt b/velox/connectors/clp/search_lib/archive/CMakeLists.txt index 3777be57d53..3fc3b92add4 100644 --- a/velox/connectors/clp/search_lib/archive/CMakeLists.txt +++ b/velox/connectors/clp/search_lib/archive/CMakeLists.txt @@ -11,9 +11,20 @@ # 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. -target_sources( - clp-s-search - PRIVATE ClpArchiveCursor.cpp ClpArchiveCursor.h ClpQueryRunner.cpp - ClpQueryRunner.h) +velox_add_library( + clp-s-archive-search + STATIC + ClpArchiveCursor.cpp + ClpArchiveCursor.h + ClpQueryRunner.cpp + ClpQueryRunner.h) -target_include_directories(clp-s-search PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}) +velox_link_libraries( + clp-s-archive-search + PUBLIC clp_s::archive_reader + PRIVATE + clp_s::clp_dependencies + clp_s::io + clp_s::search + clp_s::search::kql + velox_vector) diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h index 65859bc5b1a..e8b0c1362ae 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h @@ -31,6 +31,8 @@ namespace facebook::velox::connector::clp::search_lib { class ClpQueryRunner; +/// A query execution implementation that manages the lifecycle of a query on a +/// CLP-S archive. class ClpArchiveCursor final : public BaseClpCursor { public: explicit ClpArchiveCursor( diff --git a/velox/connectors/clp/search_lib/archive/ClpQueryRunner.cpp b/velox/connectors/clp/search_lib/archive/ClpQueryRunner.cpp index dcb6fdc4bb5..0691b0af203 100644 --- a/velox/connectors/clp/search_lib/archive/ClpQueryRunner.cpp +++ b/velox/connectors/clp/search_lib/archive/ClpQueryRunner.cpp @@ -14,7 +14,7 @@ * limitations under the License. */ -#include "ClpQueryRunner.h" +#include "velox/connectors/clp/search_lib/archive/ClpQueryRunner.h" using namespace clp_s; using namespace clp_s::search; From 19bad615bba2344bb74ce3f3e9254092a52b0d29 Mon Sep 17 00:00:00 2001 From: anlowee Date: Tue, 2 Sep 2025 19:27:13 +0000 Subject: [PATCH 06/11] Move the current VectorLoader to Archive-specific --- velox/connectors/clp/ClpDataSource.cpp | 53 +--------------- velox/connectors/clp/ClpDataSource.h | 19 ------ .../connectors/clp/search_lib/BaseClpCursor.h | 31 +++++----- .../connectors/clp/search_lib/CMakeLists.txt | 4 +- .../clp/search_lib/archive/CMakeLists.txt | 2 + .../search_lib/archive/ClpArchiveCursor.cpp | 62 +++++++++++++++++++ .../clp/search_lib/archive/ClpArchiveCursor.h | 18 +++++- .../ClpArchiveVectorLoader.cpp} | 35 ++++++----- .../ClpArchiveVectorLoader.h} | 9 +-- 9 files changed, 122 insertions(+), 111 deletions(-) rename velox/connectors/clp/search_lib/{ClpVectorLoader.cpp => archive/ClpArchiveVectorLoader.cpp} (91%) rename velox/connectors/clp/search_lib/{ClpVectorLoader.h => archive/ClpArchiveVectorLoader.h} (87%) diff --git a/velox/connectors/clp/ClpDataSource.cpp b/velox/connectors/clp/ClpDataSource.cpp index 03059f584b6..87d3d9b7172 100644 --- a/velox/connectors/clp/ClpDataSource.cpp +++ b/velox/connectors/clp/ClpDataSource.cpp @@ -16,12 +16,12 @@ #include +#include "search_lib/archive/ClpArchiveVectorLoader.h" #include "velox/connectors/clp/ClpColumnHandle.h" #include "velox/connectors/clp/ClpConnectorSplit.h" #include "velox/connectors/clp/ClpDataSource.h" #include "velox/connectors/clp/ClpTableHandle.h" #include "velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h" -#include "velox/connectors/clp/search_lib/ClpVectorLoader.h" #include "velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h" #include "velox/vector/FlatVector.h" @@ -130,46 +130,6 @@ void ClpDataSource::addSplit(std::shared_ptr split) { } } -VectorPtr ClpDataSource::createVector( - const TypePtr& vectorType, - size_t vectorSize, - const std::vector& projectedColumns, - const std::shared_ptr>& filteredRows, - size_t& readerIndex) { - if (vectorType->kind() == TypeKind::ROW) { - std::vector children; - auto& rowType = vectorType->as(); - for (uint32_t i = 0; i < rowType.size(); ++i) { - children.push_back(createVector( - rowType.childAt(i), - vectorSize, - projectedColumns, - filteredRows, - readerIndex)); - } - return std::make_shared( - pool_, vectorType, nullptr, vectorSize, std::move(children)); - } - auto vector = BaseVector::create(vectorType, vectorSize, pool_); - vector->setNulls(allocateNulls(vectorSize, pool_, bits::kNull)); - - VELOX_CHECK_LT( - readerIndex, projectedColumns.size(), "Reader index out of bounds"); - auto projectedColumn = projectedColumns[readerIndex]; - auto projectedType = fields_[readerIndex].type; - readerIndex++; - return std::make_shared( - pool_, - vectorType, - vectorSize, - std::make_unique( - projectedColumn, - projectedType, - filteredRows, - cursor_->getSplitType()), - std::move(vector)); -} - std::optional ClpDataSource::next( uint64_t size, ContinueFuture& future) { @@ -181,15 +141,8 @@ std::optional ClpDataSource::next( } completedRows_ += rowsScanned; size_t readerIndex = 0; - const auto& projectedColumns = cursor_->getProjectedColumns(); - VELOX_CHECK_EQ( - projectedColumns.size(), - fields_.size(), - "Projected columns size {} does not match fields size {}", - projectedColumns.size(), - fields_.size()); - return std::dynamic_pointer_cast(createVector( - outputType_, rowsFiltered, projectedColumns, filteredRows, readerIndex)); + return std::dynamic_pointer_cast(cursor_->createVector( + pool_, outputType_, rowsFiltered, filteredRows, readerIndex)); } } // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/ClpDataSource.h b/velox/connectors/clp/ClpDataSource.h index 32611525fe2..a500219f3f8 100644 --- a/velox/connectors/clp/ClpDataSource.h +++ b/velox/connectors/clp/ClpDataSource.h @@ -74,25 +74,6 @@ class ClpDataSource : public DataSource { const TypePtr& columnType, const std::string& parentName); - /// Creates a Vector of the specified type and size. - /// - /// This method recursively creates vectors for complex types like ROW. For - /// primitive types, it creates a LazyVector that will load the data from the - /// underlying data source when it is accessed. - /// - /// @param vectorType - /// @param vectorSize - /// @param projectedColumns The readers of the projected columns. - /// @param filteredRows The rows to be read. - /// @param readerIndex The index of the column reader. - /// @return A Vector of the specified type and size. - VectorPtr createVector( - const TypePtr& vectorType, - size_t vectorSize, - const std::vector& projectedColumns, - const std::shared_ptr>& filteredRows, - size_t& readerIndex); - ClpConfig::StorageType storageType_; velox::memory::MemoryPool* pool_; RowTypePtr outputType_; diff --git a/velox/connectors/clp/search_lib/BaseClpCursor.h b/velox/connectors/clp/search_lib/BaseClpCursor.h index 91f44a258f8..0929a4196fe 100644 --- a/velox/connectors/clp/search_lib/BaseClpCursor.h +++ b/velox/connectors/clp/search_lib/BaseClpCursor.h @@ -72,8 +72,7 @@ class BaseClpCursor { std::string_view splitPath) : errorCode_(ErrorCode::QueryNotInitialized), inputSource_(inputSource), - splitPath_(std::string(splitPath)), - splitType_(ClpConnectorSplit::SplitType::kArchive) {} + splitPath_(std::string(splitPath)) {} virtual ~BaseClpCursor() = default; /// Executes a query. This function parses, validates, and prepares the given @@ -96,19 +95,24 @@ class BaseClpCursor { uint64_t numRows, const std::shared_ptr>& filteredRowIndices) = 0; - /// Retrieves the projected columns. + /// Creates a Vector of the specified type and size. /// - /// @return A vector of BaseColumnReader pointers representing the projected - /// columns. - virtual const std::vector& getProjectedColumns() - const = 0; - - /// Get the type of the split that the cursor is processing. + /// This method recursively creates vectors for complex types like ROW. For + /// primitive types, it creates a LazyVector that will load the data from the + /// underlying data source when it is accessed. /// - /// @return The split type. - ClpConnectorSplit::SplitType getSplitType() const { - return splitType_; - } + /// @param pool The memory pool used by ClpDataSource to create the vector + /// @param vectorType + /// @param vectorSize + /// @param filteredRows The rows to be read. + /// @param readerIndex The index of the column reader. + /// @return A Vector of the specified type and size. + virtual VectorPtr createVector( + memory::MemoryPool* pool, + const TypePtr& vectorType, + size_t vectorSize, + const std::shared_ptr>& filteredRows, + size_t& readerIndex) = 0; protected: /// @@ -119,7 +123,6 @@ class BaseClpCursor { clp_s::InputSource inputSource_{clp_s::InputSource::Filesystem}; std::string splitPath_; - ClpConnectorSplit::SplitType splitType_; std::string query_; std::vector outputColumns_; diff --git a/velox/connectors/clp/search_lib/CMakeLists.txt b/velox/connectors/clp/search_lib/CMakeLists.txt index 4575eec1284..2dd05a96cf3 100644 --- a/velox/connectors/clp/search_lib/CMakeLists.txt +++ b/velox/connectors/clp/search_lib/CMakeLists.txt @@ -19,9 +19,7 @@ velox_add_library( ClpPackageS3AuthProvider.cpp ClpPackageS3AuthProvider.h ClpS3AuthProviderBase.cpp - ClpS3AuthProviderBase.h - ClpVectorLoader.cpp - ClpVectorLoader.h) + ClpS3AuthProviderBase.h) add_subdirectory(archive) diff --git a/velox/connectors/clp/search_lib/archive/CMakeLists.txt b/velox/connectors/clp/search_lib/archive/CMakeLists.txt index 3fc3b92add4..3f09cce6fd0 100644 --- a/velox/connectors/clp/search_lib/archive/CMakeLists.txt +++ b/velox/connectors/clp/search_lib/archive/CMakeLists.txt @@ -16,6 +16,8 @@ velox_add_library( STATIC ClpArchiveCursor.cpp ClpArchiveCursor.h + ClpArchiveVectorLoader.cpp + ClpArchiveVectorLoader.h ClpQueryRunner.cpp ClpQueryRunner.h) diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp index 55c5e995df2..99c6bccd208 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp @@ -22,6 +22,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/search_lib/archive/ClpArchiveVectorLoader.h" #include "velox/connectors/clp/search_lib/archive/ClpQueryRunner.h" using namespace clp_s; @@ -87,6 +88,67 @@ uint64_t ClpArchiveCursor::fetchNext( return 0; } +VectorPtr ClpArchiveCursor::createVector( + memory::MemoryPool* pool, + const TypePtr& vectorType, + size_t vectorSize, + const std::shared_ptr>& filteredRows, + size_t& readerIndex) { + auto projectedColumns = getProjectedColumns(); + VELOX_CHECK_EQ( + projectedColumns.size(), + outputColumns_.size(), + "Projected columns size {} does not match fields size {}", + projectedColumns.size(), + outputColumns_.size()); + return createVectorHelper( + pool, + vectorType, + vectorSize, + projectedColumns, + filteredRows, + readerIndex); +} + +VectorPtr ClpArchiveCursor::createVectorHelper( + memory::MemoryPool* pool, + const TypePtr& vectorType, + size_t vectorSize, + const std::vector& projectedColumns, + const std::shared_ptr>& filteredRows, + size_t& readerIndex) { + if (vectorType->kind() == TypeKind::ROW) { + std::vector children; + auto& rowType = vectorType->as(); + for (uint32_t i = 0; i < rowType.size(); ++i) { + children.push_back(createVectorHelper( + pool, + rowType.childAt(i), + vectorSize, + projectedColumns, + filteredRows, + readerIndex)); + } + return std::make_shared( + pool, vectorType, nullptr, vectorSize, std::move(children)); + } + auto vector = BaseVector::create(vectorType, vectorSize, pool); + vector->setNulls(allocateNulls(vectorSize, pool, bits::kNull)); + + VELOX_CHECK_LT( + readerIndex, projectedColumns.size(), "Reader index out of bounds"); + auto projectedColumn = projectedColumns[readerIndex]; + auto projectedType = outputColumns_[readerIndex].type; + readerIndex++; + return std::make_shared( + pool, + vectorType, + vectorSize, + std::make_unique( + projectedColumn, projectedType, filteredRows), + std::move(vector)); +} + const std::vector& ClpArchiveCursor::getProjectedColumns() const { if (queryRunner_) { diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h index e8b0c1362ae..94252028196 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h @@ -45,8 +45,12 @@ class ClpArchiveCursor final : public BaseClpCursor { const std::shared_ptr>& filteredRowIndices) override; - const std::vector& getProjectedColumns() - const override; + VectorPtr createVector( + memory::MemoryPool* pool, + const TypePtr& vectorType, + size_t vectorSize, + const std::shared_ptr>& filteredRows, + size_t& readerIndex) override; protected: ErrorCode loadSplit() override; @@ -62,6 +66,16 @@ class ClpArchiveCursor final : public BaseClpCursor { std::shared_ptr projection_; std::shared_ptr archiveReader_; + + const std::vector& getProjectedColumns() const; + + VectorPtr createVectorHelper( + memory::MemoryPool* pool, + const TypePtr& vectorType, + size_t vectorSize, + const std::vector& projectedColumns, + const std::shared_ptr>& filteredRows, + size_t& readerIndex); }; } // namespace facebook::velox::connector::clp::search_lib diff --git a/velox/connectors/clp/search_lib/ClpVectorLoader.cpp b/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp similarity index 91% rename from velox/connectors/clp/search_lib/ClpVectorLoader.cpp rename to velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp index 1471b66a5e5..29b5bbf50f8 100644 --- a/velox/connectors/clp/search_lib/ClpVectorLoader.cpp +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp @@ -21,8 +21,8 @@ #include "clp_s/ColumnReader.hpp" #include "clp_s/SchemaTree.hpp" +#include "ClpArchiveVectorLoader.h" #include "velox/connectors/clp/search_lib/BaseClpCursor.h" -#include "velox/connectors/clp/search_lib/ClpVectorLoader.h" #include "velox/type/Timestamp.h" #include "velox/vector/ComplexVector.h" #include "velox/vector/FlatVector.h" @@ -122,18 +122,16 @@ auto convertToVeloxTimestamp(int64_t timestamp) -> Timestamp { } // namespace -ClpVectorLoader::ClpVectorLoader( +ClpArchiveVectorLoader::ClpArchiveVectorLoader( clp_s::BaseColumnReader* columnReader, ColumnType nodeType, - std::shared_ptr> filteredRowIndices, - ClpConnectorSplit::SplitType splitType) + std::shared_ptr> filteredRowIndices) : columnReader_(columnReader), nodeType_(nodeType), - filteredRowIndices_(std::move(filteredRowIndices)), - splitType_(splitType) {} + filteredRowIndices_(std::move(filteredRowIndices)) {} template -void ClpVectorLoader::populateData(RowSet rows, VectorPtr vector) { +void ClpArchiveVectorLoader::populateData(RowSet rows, VectorPtr vector) { if (columnReader_ == nullptr) { for (int vectorIndex : rows) { vector->setNull(vectorIndex, true); @@ -158,7 +156,7 @@ void ClpVectorLoader::populateData(RowSet rows, VectorPtr vector) { } template -void ClpVectorLoader::populateTimestampData( +void ClpArchiveVectorLoader::populateTimestampData( RowSet rows, FlatVector* vector) { bool supportedNodeType{false}; @@ -202,7 +200,7 @@ void ClpVectorLoader::populateTimestampData( } } -void ClpVectorLoader::loadInternal( +void ClpArchiveVectorLoader::loadInternal( RowSet rows, ValueHook* hook, vector_size_t resultSize, @@ -310,29 +308,32 @@ void ClpVectorLoader::loadInternal( } // Explicit template instantiations for linker -template void ClpVectorLoader::populateData( +template void ClpArchiveVectorLoader::populateData( RowSet rows, FlatVector* vector); -template void ClpVectorLoader::populateData( +template void ClpArchiveVectorLoader::populateData( RowSet rows, FlatVector* vector); -template void ClpVectorLoader::populateData( +template void ClpArchiveVectorLoader::populateData( RowSet rows, FlatVector* vector); -template void ClpVectorLoader::populateData( +template void ClpArchiveVectorLoader::populateData( RowSet rows, FlatVector* vector); -template void ClpVectorLoader::populateTimestampData( +template void +ClpArchiveVectorLoader::populateTimestampData( RowSet rows, FlatVector* vector); -template void ClpVectorLoader::populateTimestampData( +template void +ClpArchiveVectorLoader::populateTimestampData( RowSet rows, FlatVector* vector); template void -ClpVectorLoader::populateTimestampData( +ClpArchiveVectorLoader::populateTimestampData( RowSet rows, FlatVector* vector); -template void ClpVectorLoader::populateTimestampData( +template void +ClpArchiveVectorLoader::populateTimestampData( RowSet rows, FlatVector* vector); diff --git a/velox/connectors/clp/search_lib/ClpVectorLoader.h b/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h similarity index 87% rename from velox/connectors/clp/search_lib/ClpVectorLoader.h rename to velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h index 69a7f259f24..35ad6c5ae70 100644 --- a/velox/connectors/clp/search_lib/ClpVectorLoader.h +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h @@ -20,7 +20,6 @@ #include "clp_s/ColumnReader.hpp" #include "clp_s/SchemaTree.hpp" -#include "velox/connectors/clp/ClpConnectorSplit.h" #include "velox/type/Timestamp.h" #include "velox/vector/FlatVector.h" @@ -37,13 +36,12 @@ enum class ColumnType; /// A custom Velox VectorLoader that populates Velox vectors from a CLP-based /// column reader. It supports various column types including integers, floats, /// booleans, strings, and arrays of strings. -class ClpVectorLoader : public VectorLoader { +class ClpArchiveVectorLoader : public VectorLoader { public: - ClpVectorLoader( + ClpArchiveVectorLoader( clp_s::BaseColumnReader* columnReader, ColumnType nodeType, - std::shared_ptr> filteredRowIndices, - ClpConnectorSplit::SplitType splitType); + std::shared_ptr> filteredRowIndices); private: void loadInternal( @@ -63,7 +61,6 @@ class ClpVectorLoader : public VectorLoader { clp_s::BaseColumnReader* columnReader_; ColumnType nodeType_; std::shared_ptr> filteredRowIndices_; - ClpConnectorSplit::SplitType splitType_; inline static thread_local std::unique_ptr arrayParser_ = std::make_unique(); From e466f68de088249f4c4e42a125f5c71b4ed7a525 Mon Sep 17 00:00:00 2001 From: anlowee Date: Wed, 3 Sep 2025 18:13:43 +0000 Subject: [PATCH 07/11] Put more archive specific thing into Archive code --- velox/connectors/clp/ClpDataSource.cpp | 10 ++-- .../connectors/clp/search_lib/BaseClpCursor.h | 13 ++--- .../search_lib/archive/ClpArchiveCursor.cpp | 48 ++++++++----------- .../clp/search_lib/archive/ClpArchiveCursor.h | 18 ++++--- .../archive/ClpArchiveVectorLoader.cpp | 8 ++-- .../archive/ClpArchiveVectorLoader.h | 2 +- 6 files changed, 40 insertions(+), 59 deletions(-) diff --git a/velox/connectors/clp/ClpDataSource.cpp b/velox/connectors/clp/ClpDataSource.cpp index 87d3d9b7172..7ee225be50b 100644 --- a/velox/connectors/clp/ClpDataSource.cpp +++ b/velox/connectors/clp/ClpDataSource.cpp @@ -133,16 +133,14 @@ void ClpDataSource::addSplit(std::shared_ptr split) { std::optional ClpDataSource::next( uint64_t size, ContinueFuture& future) { - auto filteredRows = std::make_shared>(); - auto rowsScanned = cursor_->fetchNext(size, filteredRows); - auto rowsFiltered = filteredRows->size(); + auto rowsScanned = cursor_->fetchNext(size); + auto rowsFiltered = cursor_->getNumFilteredRows(); if (rowsFiltered == 0) { return nullptr; } completedRows_ += rowsScanned; - size_t readerIndex = 0; - return std::dynamic_pointer_cast(cursor_->createVector( - pool_, outputType_, rowsFiltered, filteredRows, readerIndex)); + return std::dynamic_pointer_cast( + cursor_->createVector(pool_, outputType_, rowsFiltered)); } } // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/search_lib/BaseClpCursor.h b/velox/connectors/clp/search_lib/BaseClpCursor.h index 0929a4196fe..b4f40cc5340 100644 --- a/velox/connectors/clp/search_lib/BaseClpCursor.h +++ b/velox/connectors/clp/search_lib/BaseClpCursor.h @@ -89,11 +89,8 @@ class BaseClpCursor { /// loaded, this function will perform the necessary loading. /// /// @param numRows The maximum number of rows to fetch. - /// @param filteredRowIndices A vector of row indices that match the filter. /// @return The number of rows scanned. - virtual uint64_t fetchNext( - uint64_t numRows, - const std::shared_ptr>& filteredRowIndices) = 0; + virtual uint64_t fetchNext(uint64_t numRows) = 0; /// Creates a Vector of the specified type and size. /// @@ -104,15 +101,13 @@ class BaseClpCursor { /// @param pool The memory pool used by ClpDataSource to create the vector /// @param vectorType /// @param vectorSize - /// @param filteredRows The rows to be read. - /// @param readerIndex The index of the column reader. /// @return A Vector of the specified type and size. virtual VectorPtr createVector( memory::MemoryPool* pool, const TypePtr& vectorType, - size_t vectorSize, - const std::shared_ptr>& filteredRows, - size_t& readerIndex) = 0; + size_t vectorSize) = 0; + + virtual size_t getNumFilteredRows() = 0; protected: /// diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp index 99c6bccd208..e84aeed890b 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp @@ -43,9 +43,10 @@ ClpArchiveCursor::~ClpArchiveCursor() { } } -uint64_t ClpArchiveCursor::fetchNext( - uint64_t numRows, - const std::shared_ptr>& filteredRowIndices) { +uint64_t ClpArchiveCursor::fetchNext(uint64_t numRows) { + filteredRowIndices_->clear(); + readerIndex_ = 0; + if (ErrorCode::Success != errorCode_) { return 0; } @@ -76,8 +77,8 @@ uint64_t ClpArchiveCursor::fetchNext( currentSchemaTableLoaded_ = true; } - auto rowsScanned = queryRunner_->fetchNext(numRows, filteredRowIndices); - if (false == filteredRowIndices->empty()) { + auto rowsScanned = queryRunner_->fetchNext(numRows, filteredRowIndices_); + if (false == filteredRowIndices_->empty()) { return rowsScanned; } @@ -88,12 +89,14 @@ uint64_t ClpArchiveCursor::fetchNext( return 0; } +size_t ClpArchiveCursor::getNumFilteredRows() { + return filteredRowIndices_->size(); +} + VectorPtr ClpArchiveCursor::createVector( memory::MemoryPool* pool, const TypePtr& vectorType, - size_t vectorSize, - const std::shared_ptr>& filteredRows, - size_t& readerIndex) { + size_t vectorSize) { auto projectedColumns = getProjectedColumns(); VELOX_CHECK_EQ( projectedColumns.size(), @@ -101,33 +104,20 @@ VectorPtr ClpArchiveCursor::createVector( "Projected columns size {} does not match fields size {}", projectedColumns.size(), outputColumns_.size()); - return createVectorHelper( - pool, - vectorType, - vectorSize, - projectedColumns, - filteredRows, - readerIndex); + return createVectorHelper(pool, vectorType, vectorSize, projectedColumns); } VectorPtr ClpArchiveCursor::createVectorHelper( memory::MemoryPool* pool, const TypePtr& vectorType, size_t vectorSize, - const std::vector& projectedColumns, - const std::shared_ptr>& filteredRows, - size_t& readerIndex) { + 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) { children.push_back(createVectorHelper( - pool, - rowType.childAt(i), - vectorSize, - projectedColumns, - filteredRows, - readerIndex)); + pool, rowType.childAt(i), vectorSize, projectedColumns)); } return std::make_shared( pool, vectorType, nullptr, vectorSize, std::move(children)); @@ -136,16 +126,16 @@ VectorPtr ClpArchiveCursor::createVectorHelper( vector->setNulls(allocateNulls(vectorSize, pool, bits::kNull)); VELOX_CHECK_LT( - readerIndex, projectedColumns.size(), "Reader index out of bounds"); - auto projectedColumn = projectedColumns[readerIndex]; - auto projectedType = outputColumns_[readerIndex].type; - readerIndex++; + readerIndex_, projectedColumns.size(), "Reader index out of bounds"); + auto projectedColumn = projectedColumns[readerIndex_]; + auto projectedType = outputColumns_[readerIndex_].type; + readerIndex_++; return std::make_shared( pool, vectorType, vectorSize, std::make_unique( - projectedColumn, projectedType, filteredRows), + projectedColumn, projectedType, filteredRowIndices_), std::move(vector)); } diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h index 94252028196..2e098e89ef3 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h @@ -40,22 +40,22 @@ class ClpArchiveCursor final : public BaseClpCursor { std::string_view splitPath); ~ClpArchiveCursor() override; - uint64_t fetchNext( - uint64_t numRows, - const std::shared_ptr>& filteredRowIndices) - override; + uint64_t fetchNext(uint64_t numRows) override; VectorPtr createVector( memory::MemoryPool* pool, const TypePtr& vectorType, - size_t vectorSize, - const std::shared_ptr>& filteredRows, - size_t& readerIndex) override; + size_t vectorSize) override; + + size_t getNumFilteredRows() override; protected: ErrorCode loadSplit() override; private: + size_t readerIndex_{0}; + std::shared_ptr> filteredRowIndices_ = + std::make_shared>(); std::vector matchedSchemas_; size_t currentSchemaIndex_{0}; int32_t currentSchemaId_{-1}; @@ -73,9 +73,7 @@ class ClpArchiveCursor final : public BaseClpCursor { memory::MemoryPool* pool, const TypePtr& vectorType, size_t vectorSize, - const std::vector& projectedColumns, - const std::shared_ptr>& filteredRows, - size_t& readerIndex); + const std::vector& projectedColumns); }; } // namespace facebook::velox::connector::clp::search_lib diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp b/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp index 29b5bbf50f8..1ea6e6e8f69 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp @@ -125,7 +125,7 @@ auto convertToVeloxTimestamp(int64_t timestamp) -> Timestamp { ClpArchiveVectorLoader::ClpArchiveVectorLoader( clp_s::BaseColumnReader* columnReader, ColumnType nodeType, - std::shared_ptr> filteredRowIndices) + const std::shared_ptr> filteredRowIndices) : columnReader_(columnReader), nodeType_(nodeType), filteredRowIndices_(std::move(filteredRowIndices)) {} @@ -140,7 +140,7 @@ void ClpArchiveVectorLoader::populateData(RowSet rows, VectorPtr vector) { } for (int vectorIndex : rows) { - auto messageIndex = (*filteredRowIndices_)[vectorIndex]; + auto messageIndex = filteredRowIndices_->at(vectorIndex); if constexpr (std::is_same_v) { auto string_value = @@ -177,7 +177,7 @@ void ClpArchiveVectorLoader::populateTimestampData( } for (int vectorIndex : rows) { - auto messageIndex = (*filteredRowIndices_)[vectorIndex]; + auto messageIndex = filteredRowIndices_->at(vectorIndex); if (clp_s::NodeType::Float == Type) { auto reader = static_cast(columnReader_); @@ -241,7 +241,7 @@ void ClpArchiveVectorLoader::loadInternal( vector_size_t elementIndex = 0; for (int vectorIndex : rows) { - auto messageIndex = (*filteredRowIndices_)[vectorIndex]; + auto messageIndex = filteredRowIndices_->at(vectorIndex); auto jsonString = std::get(columnReader_->extract_value(messageIndex)); diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h b/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h index 35ad6c5ae70..24992b2402e 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h @@ -41,7 +41,7 @@ class ClpArchiveVectorLoader : public VectorLoader { ClpArchiveVectorLoader( clp_s::BaseColumnReader* columnReader, ColumnType nodeType, - std::shared_ptr> filteredRowIndices); + const std::shared_ptr> filteredRowIndices); private: void loadInternal( From 1c091c7ca801362f0ece3b0a4f2e7764a32dfea0 Mon Sep 17 00:00:00 2001 From: anlowee Date: Thu, 4 Sep 2025 18:59:03 +0000 Subject: [PATCH 08/11] Code clean up --- CMake/resolve_dependency_modules/clp.cmake | 2 +- .../connectors/clp/search_lib/BaseClpCursor.h | 20 +++--- .../search_lib/archive/ClpArchiveCursor.cpp | 64 +++++++++---------- .../clp/search_lib/archive/ClpArchiveCursor.h | 20 +++--- .../archive/ClpArchiveVectorLoader.cpp | 2 +- .../archive/ClpArchiveVectorLoader.h | 14 ++-- 6 files changed, 61 insertions(+), 61 deletions(-) diff --git a/CMake/resolve_dependency_modules/clp.cmake b/CMake/resolve_dependency_modules/clp.cmake index 93d57bf4da6..f2d3e5b86b5 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 19cd534e629d746395efc64343a60f768b0c9a2d) + GIT_TAG 0de99a9e8485ca3dc48710ad2cae31bffe20cd62) set(CLP_BUILD_CLP_REGEX_UTILS OFF diff --git a/velox/connectors/clp/search_lib/BaseClpCursor.h b/velox/connectors/clp/search_lib/BaseClpCursor.h index b4f40cc5340..a3f58bec2e5 100644 --- a/velox/connectors/clp/search_lib/BaseClpCursor.h +++ b/velox/connectors/clp/search_lib/BaseClpCursor.h @@ -92,6 +92,12 @@ class BaseClpCursor { /// @return The number of rows scanned. virtual uint64_t fetchNext(uint64_t numRows) = 0; + /// Gets the count of rows that satisfy the query (used to size the result + /// vector). + /// + /// @return Count of rows matching the query. + virtual size_t getNumFilteredRows() = 0; + /// Creates a Vector of the specified type and size. /// /// This method recursively creates vectors for complex types like ROW. For @@ -107,23 +113,19 @@ class BaseClpCursor { const TypePtr& vectorType, size_t vectorSize) = 0; - virtual size_t getNumFilteredRows() = 0; - protected: + /// Loads the split from archive or IR stream. /// /// @return The error code. virtual ErrorCode loadSplit() = 0; + bool currentSplitLoaded_{false}; ErrorCode errorCode_; - + std::shared_ptr expr_; clp_s::InputSource inputSource_{clp_s::InputSource::Filesystem}; - std::string splitPath_; - std::string query_; std::vector outputColumns_; - - bool currentSplitLoaded_{false}; - - std::shared_ptr expr_; + std::string query_; + std::string splitPath_; private: /// Preprocesses the query, performing parsing, validation, and optimization. diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp index 9cca7c053cd..9210eda825b 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp @@ -107,38 +107,6 @@ VectorPtr ClpArchiveCursor::createVector( return createVectorHelper(pool, vectorType, vectorSize, projectedColumns); } -VectorPtr ClpArchiveCursor::createVectorHelper( - memory::MemoryPool* pool, - const TypePtr& vectorType, - 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) { - children.push_back(createVectorHelper( - pool, rowType.childAt(i), vectorSize, projectedColumns)); - } - return std::make_shared( - pool, vectorType, nullptr, vectorSize, std::move(children)); - } - auto vector = BaseVector::create(vectorType, vectorSize, pool); - vector->setNulls(allocateNulls(vectorSize, pool, bits::kNull)); - - VELOX_CHECK_LT( - readerIndex_, projectedColumns.size(), "Reader index out of bounds"); - auto projectedColumn = projectedColumns[readerIndex_]; - auto projectedType = outputColumns_[readerIndex_].type; - readerIndex_++; - return std::make_shared( - pool, - vectorType, - vectorSize, - std::make_unique( - projectedColumn, projectedType, filteredRowIndices_), - std::move(vector)); -} - const std::vector& ClpArchiveCursor::getProjectedColumns() const { if (queryRunner_) { @@ -263,4 +231,36 @@ ErrorCode ClpArchiveCursor::loadSplit() { return ErrorCode::Success; } +VectorPtr ClpArchiveCursor::createVectorHelper( + memory::MemoryPool* pool, + const TypePtr& vectorType, + 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) { + children.push_back(createVectorHelper( + pool, rowType.childAt(i), vectorSize, projectedColumns)); + } + return std::make_shared( + pool, vectorType, nullptr, vectorSize, std::move(children)); + } + auto vector = BaseVector::create(vectorType, vectorSize, pool); + vector->setNulls(allocateNulls(vectorSize, pool, bits::kNull)); + + VELOX_CHECK_LT( + readerIndex_, projectedColumns.size(), "Reader index out of bounds"); + auto projectedColumn = projectedColumns[readerIndex_]; + auto projectedType = outputColumns_[readerIndex_].type; + readerIndex_++; + return std::make_shared( + pool, + vectorType, + vectorSize, + std::make_unique( + projectedColumn, projectedType, filteredRowIndices_), + std::move(vector)); +} + } // namespace facebook::velox::connector::clp::search_lib diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h index 2e098e89ef3..ca34837cb29 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h @@ -42,30 +42,28 @@ class ClpArchiveCursor final : public BaseClpCursor { uint64_t fetchNext(uint64_t numRows) override; + size_t getNumFilteredRows() override; + VectorPtr createVector( memory::MemoryPool* pool, const TypePtr& vectorType, size_t vectorSize) override; - size_t getNumFilteredRows() override; - protected: ErrorCode loadSplit() override; private: - size_t readerIndex_{0}; + std::shared_ptr archiveReader_; + int32_t currentSchemaId_{-1}; + size_t currentSchemaIndex_{0}; + bool currentSchemaTableLoaded_{false}; std::shared_ptr> filteredRowIndices_ = std::make_shared>(); std::vector matchedSchemas_; - size_t currentSchemaIndex_{0}; - int32_t currentSchemaId_{-1}; - bool currentSchemaTableLoaded_{false}; - - std::shared_ptr schemaMatch_; - std::shared_ptr queryRunner_; std::shared_ptr projection_; - - std::shared_ptr archiveReader_; + std::shared_ptr queryRunner_; + size_t readerIndex_{0}; + std::shared_ptr schemaMatch_; const std::vector& getProjectedColumns() const; diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp b/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp index 1ea6e6e8f69..9238674671a 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp @@ -21,8 +21,8 @@ #include "clp_s/ColumnReader.hpp" #include "clp_s/SchemaTree.hpp" -#include "ClpArchiveVectorLoader.h" #include "velox/connectors/clp/search_lib/BaseClpCursor.h" +#include "velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h" #include "velox/type/Timestamp.h" #include "velox/vector/ComplexVector.h" #include "velox/vector/FlatVector.h" diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h b/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h index 24992b2402e..171a2d7c866 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h @@ -44,6 +44,13 @@ class ClpArchiveVectorLoader : public VectorLoader { const std::shared_ptr> filteredRowIndices); private: + inline static thread_local std::unique_ptr + arrayParser_ = std::make_unique(); + + clp_s::BaseColumnReader* columnReader_; + ColumnType nodeType_; + std::shared_ptr> filteredRowIndices_; + void loadInternal( RowSet rows, ValueHook* hook, @@ -57,13 +64,6 @@ class ClpArchiveVectorLoader : public VectorLoader { void populateTimestampData( RowSet rows, FlatVector* vector); - - clp_s::BaseColumnReader* columnReader_; - ColumnType nodeType_; - std::shared_ptr> filteredRowIndices_; - - inline static thread_local std::unique_ptr - arrayParser_ = std::make_unique(); }; } // namespace facebook::velox::connector::clp::search_lib From 7f252108b71f392df70783c434ce5a003000c541 Mon Sep 17 00:00:00 2001 From: anlowee Date: Thu, 4 Sep 2025 19:47:49 +0000 Subject: [PATCH 09/11] Address coderabbitai comments --- velox/connectors/clp/ClpDataSource.cpp | 1 - velox/connectors/clp/search_lib/BaseClpCursor.cpp | 1 + velox/connectors/clp/search_lib/BaseClpCursor.h | 4 ++-- .../connectors/clp/search_lib/archive/CMakeLists.txt | 11 ++++------- .../clp/search_lib/archive/ClpArchiveCursor.cpp | 2 +- .../clp/search_lib/archive/ClpArchiveCursor.h | 2 +- 6 files changed, 9 insertions(+), 12 deletions(-) diff --git a/velox/connectors/clp/ClpDataSource.cpp b/velox/connectors/clp/ClpDataSource.cpp index 7ee225be50b..07570409111 100644 --- a/velox/connectors/clp/ClpDataSource.cpp +++ b/velox/connectors/clp/ClpDataSource.cpp @@ -16,7 +16,6 @@ #include -#include "search_lib/archive/ClpArchiveVectorLoader.h" #include "velox/connectors/clp/ClpColumnHandle.h" #include "velox/connectors/clp/ClpConnectorSplit.h" #include "velox/connectors/clp/ClpDataSource.h" diff --git a/velox/connectors/clp/search_lib/BaseClpCursor.cpp b/velox/connectors/clp/search_lib/BaseClpCursor.cpp index b0c64bf464d..2741fc58206 100644 --- a/velox/connectors/clp/search_lib/BaseClpCursor.cpp +++ b/velox/connectors/clp/search_lib/BaseClpCursor.cpp @@ -18,6 +18,7 @@ #include "velox/connectors/clp/search_lib/BaseClpCursor.h" +#include #include "clp_s/search/ast/ConvertToExists.hpp" #include "clp_s/search/ast/EmptyExpr.hpp" #include "clp_s/search/ast/NarrowTypes.hpp" diff --git a/velox/connectors/clp/search_lib/BaseClpCursor.h b/velox/connectors/clp/search_lib/BaseClpCursor.h index a3f58bec2e5..92940081f90 100644 --- a/velox/connectors/clp/search_lib/BaseClpCursor.h +++ b/velox/connectors/clp/search_lib/BaseClpCursor.h @@ -96,7 +96,7 @@ class BaseClpCursor { /// vector). /// /// @return Count of rows matching the query. - virtual size_t getNumFilteredRows() = 0; + virtual size_t getNumFilteredRows() const = 0; /// Creates a Vector of the specified type and size. /// @@ -122,7 +122,7 @@ class BaseClpCursor { bool currentSplitLoaded_{false}; ErrorCode errorCode_; std::shared_ptr expr_; - clp_s::InputSource inputSource_{clp_s::InputSource::Filesystem}; + clp_s::InputSource inputSource_; std::vector outputColumns_; std::string query_; std::string splitPath_; diff --git a/velox/connectors/clp/search_lib/archive/CMakeLists.txt b/velox/connectors/clp/search_lib/archive/CMakeLists.txt index 3f09cce6fd0..9340e22ffbb 100644 --- a/velox/connectors/clp/search_lib/archive/CMakeLists.txt +++ b/velox/connectors/clp/search_lib/archive/CMakeLists.txt @@ -23,10 +23,7 @@ velox_add_library( velox_link_libraries( clp-s-archive-search - PUBLIC clp_s::archive_reader - PRIVATE - clp_s::clp_dependencies - clp_s::io - clp_s::search - clp_s::search::kql - velox_vector) + PUBLIC clp_s::archive_reader velox_vector + PRIVATE clp_s::clp_dependencies clp_s::io clp_s::search clp_s::search::kql) + +target_compile_features(clp-s-archive-search PRIVATE cxx_std_20) diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp index 9210eda825b..615eaefeb2d 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp @@ -89,7 +89,7 @@ uint64_t ClpArchiveCursor::fetchNext(uint64_t numRows) { return 0; } -size_t ClpArchiveCursor::getNumFilteredRows() { +size_t ClpArchiveCursor::getNumFilteredRows() const { return filteredRowIndices_->size(); } diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h index ca34837cb29..6760495b158 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h @@ -42,7 +42,7 @@ class ClpArchiveCursor final : public BaseClpCursor { uint64_t fetchNext(uint64_t numRows) override; - size_t getNumFilteredRows() override; + size_t getNumFilteredRows() const override; VectorPtr createVector( memory::MemoryPool* pool, From e71a9576b986e4e12e6a93f6feb8b5c1031a7427 Mon Sep 17 00:00:00 2001 From: anlowee Date: Fri, 5 Sep 2025 14:27:43 +0000 Subject: [PATCH 10/11] Address comments --- velox/connectors/clp/search_lib/BaseClpCursor.cpp | 5 ++--- .../connectors/clp/search_lib/archive/ClpArchiveCursor.cpp | 6 +++--- velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h | 3 +-- .../clp/search_lib/archive/ClpArchiveVectorLoader.cpp | 1 - .../clp/search_lib/archive/ClpArchiveVectorLoader.h | 7 +++---- 5 files changed, 9 insertions(+), 13 deletions(-) diff --git a/velox/connectors/clp/search_lib/BaseClpCursor.cpp b/velox/connectors/clp/search_lib/BaseClpCursor.cpp index 2741fc58206..b8ab7db9457 100644 --- a/velox/connectors/clp/search_lib/BaseClpCursor.cpp +++ b/velox/connectors/clp/search_lib/BaseClpCursor.cpp @@ -14,16 +14,15 @@ * limitations under the License. */ +#include #include -#include "velox/connectors/clp/search_lib/BaseClpCursor.h" - -#include #include "clp_s/search/ast/ConvertToExists.hpp" #include "clp_s/search/ast/EmptyExpr.hpp" #include "clp_s/search/ast/NarrowTypes.hpp" #include "clp_s/search/ast/OrOfAndForm.hpp" #include "clp_s/search/kql/kql.hpp" +#include "velox/connectors/clp/search_lib/BaseClpCursor.h" using namespace clp_s; using namespace clp_s::search; diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp index 615eaefeb2d..115cc20bc61 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp @@ -16,12 +16,11 @@ #include -#include "velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h" - #include "clp_s/ArchiveReader.hpp" #include "clp_s/search/EvaluateTimestampIndex.hpp" #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/ClpArchiveVectorLoader.h" #include "velox/connectors/clp/search_lib/archive/ClpQueryRunner.h" @@ -35,7 +34,8 @@ ClpArchiveCursor::ClpArchiveCursor( clp_s::InputSource inputSource, std::string_view splitPath) : BaseClpCursor(inputSource, splitPath), - archiveReader_(std::make_shared()) {} + archiveReader_(std::make_shared()), + filteredRowIndices_(std::make_shared>()) {} ClpArchiveCursor::~ClpArchiveCursor() { if (currentSplitLoaded_) { diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h index 6760495b158..09bc042d740 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h @@ -57,8 +57,7 @@ class ClpArchiveCursor final : public BaseClpCursor { int32_t currentSchemaId_{-1}; size_t currentSchemaIndex_{0}; bool currentSchemaTableLoaded_{false}; - std::shared_ptr> filteredRowIndices_ = - std::make_shared>(); + std::shared_ptr> filteredRowIndices_; std::vector matchedSchemas_; std::shared_ptr projection_; std::shared_ptr queryRunner_; diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp b/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp index 9238674671a..1a48f9a6a91 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp @@ -20,7 +20,6 @@ #include "clp_s/ColumnReader.hpp" #include "clp_s/SchemaTree.hpp" - #include "velox/connectors/clp/search_lib/BaseClpCursor.h" #include "velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h" #include "velox/type/Timestamp.h" diff --git a/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h b/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h index 171a2d7c866..99fe5c6b505 100644 --- a/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h @@ -20,7 +20,6 @@ #include "clp_s/ColumnReader.hpp" #include "clp_s/SchemaTree.hpp" - #include "velox/type/Timestamp.h" #include "velox/vector/FlatVector.h" #include "velox/vector/LazyVector.h" @@ -33,9 +32,9 @@ namespace facebook::velox::connector::clp::search_lib { enum class ColumnType; -/// A custom Velox VectorLoader that populates Velox vectors from a CLP-based -/// column reader. It supports various column types including integers, floats, -/// booleans, strings, and arrays of strings. +/// A custom Velox VectorLoader that populates Velox vectors using a CLP-based +/// column reader over archives. It supports various column types including +/// integers, floats, booleans, strings, and arrays of strings. class ClpArchiveVectorLoader : public VectorLoader { public: ClpArchiveVectorLoader( From 37231b9a7f1584298fc2efffb9a8309a89036535 Mon Sep 17 00:00:00 2001 From: anlowee Date: Fri, 5 Sep 2025 16:09:37 +0000 Subject: [PATCH 11/11] Revert unnecessary clp version changes --- CMake/resolve_dependency_modules/clp.cmake | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CMake/resolve_dependency_modules/clp.cmake b/CMake/resolve_dependency_modules/clp.cmake index f2d3e5b86b5..93d57bf4da6 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 0de99a9e8485ca3dc48710ad2cae31bffe20cd62) + GIT_TAG 19cd534e629d746395efc64343a60f768b0c9a2d) set(CLP_BUILD_CLP_REGEX_UTILS OFF