diff --git a/velox/connectors/clp/ClpDataSource.cpp b/velox/connectors/clp/ClpDataSource.cpp index d3ad369c5b5..07570409111 100644 --- a/velox/connectors/clp/ClpDataSource.cpp +++ b/velox/connectors/clp/ClpDataSource.cpp @@ -19,11 +19,9 @@ #include "velox/connectors/clp/ClpColumnHandle.h" #include "velox/connectors/clp/ClpConnectorSplit.h" #include "velox/connectors/clp/ClpDataSource.h" - -#include "search_lib/ClpS3AuthProviderBase.h" #include "velox/connectors/clp/ClpTableHandle.h" -#include "velox/connectors/clp/search_lib/ClpCursor.h" -#include "velox/connectors/clp/search_lib/ClpVectorLoader.h" +#include "velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h" +#include "velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h" #include "velox/vector/FlatVector.h" namespace facebook::velox::connector::clp { @@ -104,13 +102,23 @@ 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, - s3AuthProvider_->constructS3Url(clpSplit->path_)); + std::string splitPath = 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; + splitPath = s3AuthProvider_->constructS3Url(clpSplit->path_); + } else { + VELOX_UNREACHABLE(); + } + + if (ClpConnectorSplit::SplitType::kArchive == clpSplit->type_) { + cursor_ = + std::make_unique(inputSource, splitPath); + } else { + VELOX_UNSUPPORTED( + "Unsupported split type: {}", static_cast(clpSplit->type_)); } auto pushDownQuery = clpSplit->kqlQuery_; @@ -121,63 +129,17 @@ 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), - std::move(vector)); -} - 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; - 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)); } } // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/ClpDataSource.h b/velox/connectors/clp/ClpDataSource.h index 70ec4564354..a500219f3f8 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; @@ -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_; @@ -102,7 +83,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..b8ab7db9457 --- /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 +#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; +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 55% rename from velox/connectors/clp/search_lib/ClpCursor.h rename to velox/connectors/clp/search_lib/BaseClpCursor.h index f6303a2ef94..92940081f90 100644 --- a/velox/connectors/clp/search_lib/ClpCursor.h +++ b/velox/connectors/clp/search_lib/BaseClpCursor.h @@ -16,22 +16,18 @@ #pragma once +#include #include +#include #include -#include "velox/connectors/clp/search_lib/ClpQueryRunner.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 @@ -65,14 +61,19 @@ 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 ClpCursor { +class BaseClpCursor { public: - explicit ClpCursor(clp_s::InputSource inputSource, std::string archivePath); - ~ClpCursor(); + explicit BaseClpCursor( + clp_s::InputSource inputSource, + std::string_view splitPath) + : errorCode_(ErrorCode::QueryNotInitialized), + inputSource_(inputSource), + splitPath_(std::string(splitPath)) {} + virtual ~BaseClpCursor() = default; /// Executes a query. This function parses, validates, and prepares the given /// query for execution. @@ -84,50 +85,53 @@ class ClpCursor { const std::string& query, const std::vector& outputColumns); - /// Fetches the next set of rows from the cursor. If the archive 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. /// @return The number of rows scanned. - uint64_t fetchNext( - uint64_t numRows, - const std::shared_ptr>& filteredRowIndices); + virtual uint64_t fetchNext(uint64_t numRows) = 0; - /// Retrieves the projected columns. + /// Gets the count of rows that satisfy the query (used to size the result + /// vector). /// - /// @return A vector of BaseColumnReader pointers representing the projected - /// columns. - const std::vector& getProjectedColumns() const; + /// @return Count of rows matching the query. + virtual size_t getNumFilteredRows() const = 0; - private: - /// Preprocesses the query, performing parsing, validation, and optimization. + /// Creates a Vector of the specified type and size. /// - /// @return The error code. - ErrorCode preprocessQuery(); - - /// Loads the archive at the current index. + /// 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 pool The memory pool used by ClpDataSource to create the vector + /// @param vectorType + /// @param vectorSize + /// @return A Vector of the specified type and size. + virtual VectorPtr createVector( + memory::MemoryPool* pool, + const TypePtr& vectorType, + size_t vectorSize) = 0; + + protected: + /// Loads the split from archive or IR stream. /// /// @return The error code. - ErrorCode loadArchive(); + virtual ErrorCode loadSplit() = 0; + bool currentSplitLoaded_{false}; ErrorCode errorCode_; - - clp_s::InputSource inputSource_{clp_s::InputSource::Filesystem}; - std::string archivePath_; - std::string query_; + std::shared_ptr expr_; + clp_s::InputSource inputSource_; std::vector outputColumns_; - std::vector matchedSchemas_; - size_t currentSchemaIndex_{0}; - int32_t currentSchemaId_{-1}; - bool currentSchemaTableLoaded_{false}; - bool currentArchiveLoaded_{false}; + std::string query_; + std::string splitPath_; - std::shared_ptr expr_; - std::shared_ptr schemaMatch_; - std::shared_ptr queryRunner_; - std::shared_ptr projection_; - std::shared_ptr archiveReader_; + 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 8cd5a414d7a..2dd05a96cf3 100644 --- a/velox/connectors/clp/search_lib/CMakeLists.txt +++ b/velox/connectors/clp/search_lib/CMakeLists.txt @@ -14,24 +14,15 @@ velox_add_library( clp-s-search STATIC - ClpCursor.cpp - ClpCursor.h + BaseClpCursor.cpp + BaseClpCursor.h ClpPackageS3AuthProvider.cpp ClpPackageS3AuthProvider.h - ClpQueryRunner.cpp - ClpQueryRunner.h ClpS3AuthProviderBase.cpp - ClpS3AuthProviderBase.h - ClpVectorLoader.cpp - ClpVectorLoader.h) + ClpS3AuthProviderBase.h) + +add_subdirectory(archive) + +velox_link_libraries(clp-s-search PUBLIC clp-s-archive-search) -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) 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 new file mode 100644 index 00000000000..9340e22ffbb --- /dev/null +++ b/velox/connectors/clp/search_lib/archive/CMakeLists.txt @@ -0,0 +1,29 @@ +# 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. +velox_add_library( + clp-s-archive-search + STATIC + ClpArchiveCursor.cpp + ClpArchiveCursor.h + ClpArchiveVectorLoader.cpp + ClpArchiveVectorLoader.h + ClpQueryRunner.cpp + ClpQueryRunner.h) + +velox_link_libraries( + clp-s-archive-search + 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/ClpCursor.cpp b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp similarity index 66% rename from velox/connectors/clp/search_lib/ClpCursor.cpp rename to velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp index 05cd5b32093..115cc20bc61 100644 --- a/velox/connectors/clp/search_lib/ClpCursor.cpp +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp @@ -18,14 +18,11 @@ #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" - -#include "velox/connectors/clp/search_lib/ClpCursor.h" +#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" using namespace clp_s; using namespace clp_s::search; @@ -33,44 +30,32 @@ 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)), - archiveReader_(std::make_shared()) {} +ClpArchiveCursor::ClpArchiveCursor( + clp_s::InputSource inputSource, + std::string_view splitPath) + : BaseClpCursor(inputSource, splitPath), + archiveReader_(std::make_shared()), + filteredRowIndices_(std::make_shared>()) {} -ClpCursor::~ClpCursor() { - if (currentArchiveLoaded_) { +ClpArchiveCursor::~ClpArchiveCursor() { + if (currentSplitLoaded_) { archiveReader_->close(); } } -void ClpCursor::executeQuery( - const std::string& query, - const std::vector& outputColumns) { - query_ = query; - outputColumns_ = outputColumns; - errorCode_ = preprocessQuery(); -} +uint64_t ClpArchiveCursor::fetchNext(uint64_t numRows) { + filteredRowIndices_->clear(); + readerIndex_ = 0; -uint64_t ClpCursor::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; - queryRunner_ = std::make_shared( - schemaMatch_, expr_, archiveReader_, false, projection_); - queryRunner_->global_init(); } while (currentSchemaIndex_ < matchedSchemas_.size()) { @@ -92,8 +77,8 @@ uint64_t ClpCursor::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; } @@ -104,8 +89,26 @@ uint64_t ClpCursor::fetchNext( return 0; } -const std::vector& ClpCursor::getProjectedColumns() - const { +size_t ClpArchiveCursor::getNumFilteredRows() const { + return filteredRowIndices_->size(); +} + +VectorPtr ClpArchiveCursor::createVector( + memory::MemoryPool* pool, + const TypePtr& vectorType, + size_t vectorSize) { + 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); +} + +const std::vector& +ClpArchiveCursor::getProjectedColumns() const { if (queryRunner_) { return queryRunner_->getProjectedColumns(); } @@ -113,51 +116,14 @@ const std::vector& ClpCursor::getProjectedColumns() return kEmpty; } -ErrorCode ClpCursor::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 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; @@ -256,7 +222,45 @@ ErrorCode ClpCursor::loadArchive() { 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; } +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 new file mode 100644 index 00000000000..09bc042d740 --- /dev/null +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h @@ -0,0 +1,76 @@ +/* + * 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/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; + +/// A query execution implementation that manages the lifecycle of a query on a +/// CLP-S archive. +class ClpArchiveCursor final : public BaseClpCursor { + public: + explicit ClpArchiveCursor( + clp_s::InputSource inputSource, + std::string_view splitPath); + ~ClpArchiveCursor() override; + + uint64_t fetchNext(uint64_t numRows) override; + + size_t getNumFilteredRows() const override; + + VectorPtr createVector( + memory::MemoryPool* pool, + const TypePtr& vectorType, + size_t vectorSize) override; + + protected: + ErrorCode loadSplit() override; + + private: + std::shared_ptr archiveReader_; + int32_t currentSchemaId_{-1}; + size_t currentSchemaIndex_{0}; + bool currentSchemaTableLoaded_{false}; + std::shared_ptr> filteredRowIndices_; + std::vector matchedSchemas_; + std::shared_ptr projection_; + std::shared_ptr queryRunner_; + size_t readerIndex_{0}; + std::shared_ptr schemaMatch_; + + const std::vector& getProjectedColumns() const; + + VectorPtr createVectorHelper( + memory::MemoryPool* pool, + const TypePtr& vectorType, + size_t vectorSize, + const std::vector& projectedColumns); +}; + +} // 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 89% rename from velox/connectors/clp/search_lib/ClpVectorLoader.cpp rename to velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp index 6bd58e8a2a2..1a48f9a6a91 100644 --- a/velox/connectors/clp/search_lib/ClpVectorLoader.cpp +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp @@ -20,8 +20,8 @@ #include "clp_s/ColumnReader.hpp" #include "clp_s/SchemaTree.hpp" - -#include "velox/connectors/clp/search_lib/ClpVectorLoader.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" @@ -121,16 +121,16 @@ auto convertToVeloxTimestamp(int64_t timestamp) -> Timestamp { } // namespace -ClpVectorLoader::ClpVectorLoader( +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)) {} 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); @@ -139,7 +139,7 @@ void ClpVectorLoader::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 = @@ -155,7 +155,7 @@ void ClpVectorLoader::populateData(RowSet rows, VectorPtr vector) { } template -void ClpVectorLoader::populateTimestampData( +void ClpArchiveVectorLoader::populateTimestampData( RowSet rows, FlatVector* vector) { bool supportedNodeType{false}; @@ -176,7 +176,7 @@ void ClpVectorLoader::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_); @@ -199,7 +199,7 @@ void ClpVectorLoader::populateTimestampData( } } -void ClpVectorLoader::loadInternal( +void ClpArchiveVectorLoader::loadInternal( RowSet rows, ValueHook* hook, vector_size_t resultSize, @@ -240,7 +240,7 @@ void ClpVectorLoader::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)); @@ -307,29 +307,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 80% rename from velox/connectors/clp/search_lib/ClpVectorLoader.h rename to velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h index 36af6d7b807..99fe5c6b505 100644 --- a/velox/connectors/clp/search_lib/ClpVectorLoader.h +++ b/velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h @@ -16,10 +16,10 @@ #pragma once +#include + #include "clp_s/ColumnReader.hpp" #include "clp_s/SchemaTree.hpp" - -#include "velox/connectors/clp/search_lib/ClpCursor.h" #include "velox/type/Timestamp.h" #include "velox/vector/FlatVector.h" #include "velox/vector/LazyVector.h" @@ -30,17 +30,26 @@ class BaseColumnReader; namespace facebook::velox::connector::clp::search_lib { -/// 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 { +enum class ColumnType; + +/// 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: - ClpVectorLoader( + ClpArchiveVectorLoader( clp_s::BaseColumnReader* columnReader, ColumnType nodeType, - std::shared_ptr> filteredRowIndices); + 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, @@ -54,13 +63,6 @@ class ClpVectorLoader : 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 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 300606a9007..e61b9a04d55 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 "velox/connectors/clp/search_lib/archive/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