feat: Implement IR cursor to allow searches over IR streams.#29
Conversation
…ementation to ClpArchiveCursor
WalkthroughUpdated CLP dependency tag and added ystdlib include. Refactored ClpDataSource to instantiate type-specific cursors (archive vs IR) and changed vector creation/fetch flow. Reorganized search_lib to introduce BaseClpCursor and an IR sublibrary (cursor, unit handler, vector loader). Added IR output assertions to tests. Changes
Sequence Diagram(s)sequenceDiagram
autonumber
actor QueryEngine
participant DataSource as ClpDataSource
participant Cursor as ClpArchiveCursor/ClpIrCursor
participant Storage as Filesystem/S3
rect rgb(245,250,255)
note over QueryEngine,DataSource: Split registration
QueryEngine->>DataSource: addSplit(split)
DataSource->>DataSource: determine inputSource & splitPath
alt splitType == Archive
DataSource->>Cursor: instantiate ClpArchiveCursor(inputSource, splitPath, ignoreCase)
else splitType == IR
DataSource->>Cursor: instantiate ClpIrCursor(inputSource, splitPath, ignoreCase)
else
DataSource->>DataSource: VELOX_UNREACHABLE / VELOX_UNSUPPORTED
end
end
rect rgb(242,255,242)
note over QueryEngine,Cursor: Fetch cycle
QueryEngine->>DataSource: next(maxSize)
DataSource->>Cursor: fetchNext(maxSize)
Cursor->>Storage: loadSplit if needed
Cursor-->>DataSource: rowsRead
DataSource->>Cursor: getNumFilteredRows()
alt filteredRows == 0
DataSource-->>QueryEngine: nullptr (no rows)
else
DataSource->>Cursor: createVector(pool, outputType, rowsFiltered)
Cursor-->>DataSource: VectorPtr
DataSource-->>QueryEngine: VectorPtr
end
end
sequenceDiagram
autonumber
participant ClpIrCursor
participant IRReader as IR Reader
participant Deser as IR Deserializer
participant Handler as ClpIrUnitHandler
participant Loader as ClpIrVectorLoader / LazyVector
note over ClpIrCursor,Handler: IR fetch & lazy projection
ClpIrCursor->>IRReader: open(splitPath, auth)
ClpIrCursor->>Deser: initialize(deserializer with Handler & projections)
loop up to numRows
ClpIrCursor->>Deser: deserialize_next_ir_unit()
Deser->>Handler: handle_log_event(...)
end
ClpIrCursor->>Handler: getFilteredLogEvents()
ClpIrCursor->>Loader: create loader(nodeType, nodeId, filteredEvents)
ClpIrCursor-->>ClpIrCursor: create LazyVector via createVectorHelper
Estimated code review effort🎯 4 (Complex) | ⏱️ ~75 minutes Pre-merge checks (2 passed, 1 warning)❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✨ Finishing Touches
🧪 Generate unit tests
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
@coderabbitai Please just review the code under velox/connectors/clp/search_lib/ir and unittests. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (5)
velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp (3)
17-21: Missing include for std::max.std::max is used (Line 271) but isn’t included. Add it to avoid reliance on transitive includes.
Apply:
#include <cmath> #include <cstdint> #include <utility> +#include <algorithm>
73-91: Improve timestamp splitting for doubles (rounding edge-cases).Use integer rounding for nanos and normalise the carry to avoid 0.999999999 → 1s off-by-one due to FP error.
auto convertToVeloxTimestamp(double timestamp) -> Timestamp { switch (estimatePrecision(timestamp)) { @@ } - double seconds{std::floor(timestamp)}; - double nanoseconds{(timestamp - seconds) * Timestamp::kNanosInSecond}; - return Timestamp( - static_cast<int64_t>(seconds), static_cast<uint64_t>(nanoseconds)); + const double sec_d = std::floor(timestamp); + int64_t seconds = static_cast<int64_t>(sec_d); + int64_t nanos = static_cast<int64_t>( + std::llround((timestamp - sec_d) * Timestamp::kNanosInSecond)); + if (nanos == Timestamp::kNanosInSecond) { + ++seconds; + nanos = 0; + } else if (nanos < 0) { + --seconds; + nanos += Timestamp::kNanosInSecond; + } + return Timestamp(seconds, static_cast<uint64_t>(nanos)); }
242-284: Avoid double iteration and temporary allocations when materialising arrays.Stream elements directly into the elements vector and set offset/size from the starting index. This reduces allocations and copies.
- std::vector<std::string_view> arrayElements; - for (auto arrayElement : array) { - arrayElements.emplace_back( - simdjson::to_json_string(arrayElement).value()); - } - - if (elementIndex + arrayElements.size() > elements->size()) { - size_t newSize = std::max<size_t>( - elementIndex + arrayElements.size(), - static_cast<size_t>(elements->size()) * 2); - elements->resize(newSize); - } - - arrayVector->setOffsetAndSize( - vectorIndex, elementIndex, arrayElements.size()); - for (auto& arrayElement : arrayElements) { - elements->set(elementIndex++, StringView(arrayElement)); - } + const auto start = elementIndex; + for (auto arrayElement : array) { + auto sv = simdjson::to_json_string(arrayElement).value(); + if (elementIndex >= elements->size()) { + size_t newSize = std::max<size_t>( + static_cast<size_t>(elements->size()) * 2, + static_cast<size_t>(elementIndex + 1)); + elements->resize(newSize); + } + elements->set(elementIndex++, StringView(sv)); + } + arrayVector->setOffsetAndSize(vectorIndex, start, elementIndex - start);velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp (2)
64-70: Don’t persist non-fatal errors across schemas; breaks subsequent fetches.Setting errorCode_ to DictionaryNotFound here causes later fetchNext() calls to short-circuit even if later schemas succeed.
Apply:
- if (EvaluatedValue::False == - queryRunner_->schema_init(currentSchemaId_)) { + if (EvaluatedValue::False == + queryRunner_->schema_init(currentSchemaId_)) { currentSchemaIndex_ += 1; currentSchemaTableLoaded_ = false; - errorCode_ = ErrorCode::DictionaryNotFound; continue; }
213-217: Duplicate timestamp-range evaluation.Timestamp index is evaluated twice (earlier at Lines 136–140). Unless expr_ changes in a way that invalidates the first result, drop this second check or document why it’s necessary.
Do you want me to remove the second block and update tests if any rely on this?
♻️ Duplicate comments (13)
CMake/resolve_dependency_modules/clp.cmake (1)
40-44: Guard and prepend module path; fail fast if CLP modules move.Prevents silent misconfiguration and matches earlier guidance.
-FetchContent_Populate(clp) - -add_subdirectory(${clp_SOURCE_DIR}/components/core - ${clp_BINARY_DIR}/components/core) +FetchContent_Populate(clp) + +if (EXISTS "${clp_SOURCE_DIR}/components/core/cmake/Modules") + list(PREPEND CMAKE_MODULE_PATH "${clp_SOURCE_DIR}/components/core/cmake/Modules") +else() + message(FATAL_ERROR "CLP core CMake modules not found at: ${clp_SOURCE_DIR}/components/core/cmake/Modules") +endif() +add_subdirectory(${clp_SOURCE_DIR}/components/core + ${clp_BINARY_DIR}/components/core)velox/connectors/clp/search_lib/BaseClpCursor.cpp (4)
33-39: Return the preprocessing status from executeQuery.Avoids silent failures at call sites; let callers act on the status.
-void BaseClpCursor::executeQuery( +ErrorCode BaseClpCursor::executeQuery( const std::string& query, const std::vector<Field>& outputColumns) { query_ = query; outputColumns_ = outputColumns; - errorCode_ = preprocessQuery(); + return (errorCode_ = preprocessQuery()); }Note: update the declaration and call sites accordingly (and keep errorCode_ assignment if you rely on it elsewhere).
54-59: Split assign-and-test for readability.Make the pass and emptiness check explicit.
- OrOfAndForm standardizePass; - if (expr_ = standardizePass.run(expr_); - std::dynamic_pointer_cast<EmptyExpr>(expr_)) { + OrOfAndForm standardizePass; + expr_ = standardizePass.run(expr_); + if (std::dynamic_pointer_cast<EmptyExpr>(expr_)) { VLOG(2) << "Query '" << query_ << "' is logically false"; return ErrorCode::LogicalError; }
61-66: Apply the same clarity tweak to NarrowTypes.- NarrowTypes narrowPass; - if (expr_ = narrowPass.run(expr_); - std::dynamic_pointer_cast<EmptyExpr>(expr_)) { + NarrowTypes narrowPass; + expr_ = narrowPass.run(expr_); + if (std::dynamic_pointer_cast<EmptyExpr>(expr_)) { VLOG(2) << "Query '" << query_ << "' is logically false"; return ErrorCode::LogicalError; }
68-73: Apply the same clarity tweak to ConvertToExists.- ConvertToExists convertPass; - if (expr_ = convertPass.run(expr_); - std::dynamic_pointer_cast<EmptyExpr>(expr_)) { + ConvertToExists convertPass; + expr_ = convertPass.run(expr_); + if (std::dynamic_pointer_cast<EmptyExpr>(expr_)) { VLOG(2) << "Query '" << query_ << "' is logically false"; return ErrorCode::LogicalError; }velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h (3)
40-44: Pass shared_ptr by const reference; current std::move is a no-op.The ctor takes a const shared_ptr by value, so std::move copies. Pass by const ref (or take by value and std::move into the member but drop const).
ClpArchiveVectorLoader( clp_s::BaseColumnReader* columnReader, ColumnType nodeType, - const std::shared_ptr<std::vector<uint64_t>> filteredRowIndices); + const std::shared_ptr<std::vector<uint64_t>>& filteredRowIndices);And in ClpArchiveVectorLoader.cpp:
- const std::shared_ptr<std::vector<uint64_t>> filteredRowIndices) + const std::shared_ptr<std::vector<uint64_t>>& filteredRowIndices) @@ - filteredRowIndices_(std::move(filteredRowIndices)) {} + filteredRowIndices_(filteredRowIndices) {}
46-48: Use by-value thread_local parser instead of heap allocation.Avoids an extra allocation and pointer indirection; simpler lifetime.
- inline static thread_local std::unique_ptr<simdjson::ondemand::parser> - arrayParser_ = std::make_unique<simdjson::ondemand::parser>(); + inline static thread_local simdjson::ondemand::parser arrayParser_{};Also update .cpp uses from
arrayParser_->toarrayParser_..
33-34: Invalid forward declaration of enum class; include the real definition.Forward-declaring an enum class without an explicit underlying type is ill-formed; also risks ABI mismatches. Include BaseClpCursor.h and remove the forward declaration. (Per retrieved learning from PR #25.)
-#include <simdjson.h> +#include <simdjson.h> +#include "velox/connectors/clp/search_lib/BaseClpCursor.h" @@ -enum class ColumnType;velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp (1)
124-131: Ineffective move of shared_ptr due to const parameter.std::move on a const std::shared_ptr copies. Drop const to enable move, or take by const& and copy explicitly. Also update the header.
-ClpArchiveVectorLoader::ClpArchiveVectorLoader( - clp_s::BaseColumnReader* columnReader, - ColumnType nodeType, - const std::shared_ptr<std::vector<uint64_t>> filteredRowIndices) +ClpArchiveVectorLoader::ClpArchiveVectorLoader( + clp_s::BaseColumnReader* columnReader, + ColumnType nodeType, + std::shared_ptr<std::vector<uint64_t>> filteredRowIndices) : columnReader_(columnReader), nodeType_(nodeType), filteredRowIndices_(std::move(filteredRowIndices)) {}And in ClpArchiveVectorLoader.h:
- const std::shared_ptr<std::vector<uint64_t>> filteredRowIndices); + std::shared_ptr<std::vector<uint64_t>> filteredRowIndices);velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h (2)
19-19: Add required standard headers; avoid transitive includes.This header uses std::string_view, std::vector, and std::shared_ptr. Include their headers directly.
#include "velox/connectors/clp/search_lib/BaseClpCursor.h" + +#include <memory> +#include <string_view> +#include <vector>
21-23: Forward-declare clp_s::BaseColumnReader used in signatures.getProjectedColumns() and createVectorHelper() expose BaseColumnReader*, but the type isn’t declared here.
namespace clp_s { class ArchiveReader; +class BaseColumnReader; } // namespace clp_svelox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp (2)
96-108: Validate projected/field alignment earlier (only once).Move this VELOX_CHECK_EQ to loadSplit() after projection is fully resolved and cache the size; avoid re-checking on every createVector().
Would you like me to draft a minimal patch that caches projectedColumns.size() and asserts once?
252-263: Avoid creating a loader when the projected column is null.Return the pre-null-initialised vector in that case; saves allocations.
auto projectedColumn = projectedColumns[readerIndex_]; auto projectedType = outputColumns_[readerIndex_].type; readerIndex_++; - return std::make_shared<LazyVector>( + if (projectedColumn == nullptr) { + return vector; // already null-initialised + } + return std::make_shared<LazyVector>( pool, vectorType, vectorSize, std::make_unique<ClpArchiveVectorLoader>( projectedColumn, projectedType, filteredRowIndices_), std::move(vector));
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (6)
CMake/resolve_dependency_modules/clp.cmake(1 hunks)velox/connectors/clp/search_lib/BaseClpCursor.cpp(1 hunks)velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp(4 hunks)velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h(1 hunks)velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp(8 hunks)velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h(2 hunks)
🧰 Additional context used
🧠 Learnings (4)
📓 Common learnings
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h:36-39
Timestamp: 2025-08-20T18:00:51.241Z
Learning: In ClpArchiveCursor.h, the clp_s::InputSource type is available through transitive inclusion via BaseClpCursor.h, which includes "clp_s/InputConfig.hpp" where InputSource is defined. The constructor parameter type is properly accessible without needing direct inclusion.
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/ClpVectorLoader.cpp:24-24
Timestamp: 2025-08-20T18:13:32.442Z
Learning: In ClpVectorLoader.cpp, the include of BaseClpCursor.h is necessary because the implementation uses ColumnType enum values (like ColumnType::Integer, ColumnType::Float) in a switch statement. While ClpVectorLoader.h has a forward declaration of ColumnType, the full definition from BaseClpCursor.h is required in the implementation file to access the enum values.
📚 Learning: 2025-08-20T18:00:51.241Z
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h:36-39
Timestamp: 2025-08-20T18:00:51.241Z
Learning: In ClpArchiveCursor.h, the clp_s::InputSource type is available through transitive inclusion via BaseClpCursor.h, which includes "clp_s/InputConfig.hpp" where InputSource is defined. The constructor parameter type is properly accessible without needing direct inclusion.
Applied to files:
velox/connectors/clp/search_lib/archive/ClpArchiveCursor.hvelox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cppvelox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp
📚 Learning: 2025-08-20T18:13:32.442Z
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/ClpVectorLoader.cpp:24-24
Timestamp: 2025-08-20T18:13:32.442Z
Learning: In ClpVectorLoader.cpp, the include of BaseClpCursor.h is necessary because the implementation uses ColumnType enum values (like ColumnType::Integer, ColumnType::Float) in a switch statement. While ClpVectorLoader.h has a forward declaration of ColumnType, the full definition from BaseClpCursor.h is required in the implementation file to access the enum values.
Applied to files:
velox/connectors/clp/search_lib/archive/ClpArchiveCursor.hvelox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cppvelox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h
📚 Learning: 2025-08-20T16:58:56.256Z
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp:32-35
Timestamp: 2025-08-20T16:58:56.256Z
Learning: In the BaseClpCursor class, the constructor takes std::string_view splitPath but stores it as an owning std::string member (splitPath_) using explicit conversion: splitPath_(std::string(splitPath)). This pattern safely avoids dangling reference issues when accepting string_view parameters.
Applied to files:
velox/connectors/clp/search_lib/archive/ClpArchiveCursor.hvelox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp
🧬 Code graph analysis (5)
velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h (5)
velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h (2)
clp_s(27-29)velox/connectors/clp/search_lib/BaseClpCursor.h (3)
clp_s(27-29)clp_s(31-33)BaseClpCursor(68-137)velox/connectors/clp/search_lib/archive/ClpQueryRunner.h (6)
clp_s(21-26)clp_s(28-31)clp_s(33-35)ClpQueryRunner(43-88)vector(76-78)velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp (14)
ClpArchiveCursor(33-38)ClpArchiveCursor(40-44)fetchNext(46-90)fetchNext(46-46)getNumFilteredRows(92-94)getNumFilteredRows(92-92)createVector(96-108)createVector(96-99)loadSplit(119-232)loadSplit(119-119)getProjectedColumns(110-117)getProjectedColumns(111-111)createVectorHelper(234-264)createVectorHelper(234-238)velox/connectors/clp/search_lib/archive/ClpQueryRunner.cpp (2)
fetchNext(60-78)fetchNext(60-62)
velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp (1)
velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h (1)
ClpArchiveVectorLoader(38-66)
velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.h (2)
velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp (1)
ClpArchiveVectorLoader(124-130)velox/vector/LazyVector.h (1)
VectorLoader(177-210)
velox/connectors/clp/search_lib/BaseClpCursor.cpp (1)
velox/connectors/clp/search_lib/BaseClpCursor.h (2)
clp_s(27-29)clp_s(31-33)
velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp (4)
velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h (4)
clp_s(21-23)clp_s(25-28)readerIndex_(64-64)getNumFilteredRows(45-58)velox/connectors/clp/search_lib/BaseClpCursor.h (3)
clp_s(27-29)clp_s(31-33)currentSplitLoaded_(122-135)velox/connectors/clp/search_lib/archive/ClpQueryRunner.h (4)
clp_s(21-26)clp_s(28-31)clp_s(33-35)vector(76-78)velox/connectors/clp/search_lib/archive/ClpQueryRunner.cpp (2)
fetchNext(60-78)fetchNext(60-62)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: Build with GCC / Ubuntu release with resolve_dependency
🔇 Additional comments (2)
velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp (2)
23-24: Correct include for ColumnType usage.Including BaseClpCursor.h here is necessary to use ColumnType values in the switch. Looks good.
145-151: String lifetime assumptions. Verify FlatVector::set copies.If set() ever becomes zero-copy, using a temporary std::string risks dangling references. Confirm set() copies into the vector's string buffer.
Would you like me to add a defensive comment here stating the copy semantics?
| GIT_REPOSITORY https://github.com/LinZhihao-723/clp.git | ||
| GIT_TAG f5922f5eb292784a865872b810399da7e7b94a52) |
There was a problem hiding this comment.
💡 Verification agent
❓ Verification inconclusive
Fork pin: confirm provenance and plan to revert to upstream.
Switching to a personal fork increases supply-chain risk. Please confirm the forked commit contains only the required changes and capture a plan/TODO to return to upstream y-scope/clp once available. Consider a cache var to override the repo at configure time.
Apply to make repo configurable and keep the hard pin:
FetchContent_Declare(
clp
- GIT_REPOSITORY https://github.com/LinZhihao-723/clp.git
- GIT_TAG f5922f5eb292784a865872b810399da7e7b94a52)
+ GIT_REPOSITORY ${CLP_GIT_REPOSITORY_OVERRIDE}
+ GIT_TAG f5922f5eb292784a865872b810399da7e7b94a52
+ GIT_SHALLOW TRUE
+ GIT_PROGRESS TRUE)
+
+set(CLP_GIT_REPOSITORY_OVERRIDE
+ "https://github.com/LinZhihao-723/clp.git"
+ CACHE STRING "Override CLP git repository URL (temporary until upstream catches up)")Make CLP repository URL configurable and document revert plan
- Confirm that commit f5922f5eb292784a865872b810399da7e7b94a52 in your fork (LinZhihao-723/clp) contains only the intended changes and add a TODO to switch back to upstream y-scope/clp once that catch-up commit is merged.
- Introduce a
CLP_GIT_REPOSITORY_OVERRIDECMake CACHE STRING (defaulting to your fork URL) inFetchContent_Declare, preserve the hard-pinned tag, and addGIT_SHALLOW TRUEandGIT_PROGRESS TRUEto the declaration.
🤖 Prompt for AI Agents
In CMake/resolve_dependency_modules/clp.cmake around lines 18-19, the CLP repo
URL is hard-coded to your fork and needs to be made configurable and documented;
add a CMake CACHE STRING variable named CLP_GIT_REPOSITORY_OVERRIDE (default
value set to https://github.com/LinZhihao-723/clp.git) and use that variable in
the FetchContent_Declare GIT_REPOSITORY field, keep the existing GIT_TAG
f5922f5eb292784a865872b810399da7e7b94a52 unchanged, add GIT_SHALLOW TRUE and
GIT_PROGRESS TRUE to the declaration, and add a TODO comment noting to revert
the override to the upstream y-scope/clp once the catch-up commit is merged
(also verify that the pinned commit in your fork contains only the intended
changes before shipping).
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (2)
CMake/resolve_dependency_modules/clp.cmake (2)
16-20: Enable shallow clone and progress to speed up CI fetchesKeeps determinism while cutting fetch time and surfacing progress in verbose CI logs.
FetchContent_Declare( clp GIT_REPOSITORY https://github.com/y-scope/clp.git - GIT_TAG bfd4f60ffe9c5d69618cc8416ec6729c76ee9862) + GIT_TAG bfd4f60ffe9c5d69618cc8416ec6729c76ee9862 + GIT_SHALLOW TRUE + GIT_PROGRESS TRUE)
18-19: Make CLP repository URL overridable (default to upstream)Eases temporary forks/hotfix testing without editing the file; keeps upstream as the default.
-FetchContent_Declare( - clp - GIT_REPOSITORY https://github.com/y-scope/clp.git +set(CLP_GIT_REPOSITORY_OVERRIDE + "https://github.com/y-scope/clp.git" + CACHE STRING "Override CLP git repository URL (advanced)") + +FetchContent_Declare( + clp + GIT_REPOSITORY ${CLP_GIT_REPOSITORY_OVERRIDE} GIT_TAG bfd4f60ffe9c5d69618cc8416ec6729c76ee9862 GIT_SHALLOW TRUE GIT_PROGRESS TRUE)
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (1)
CMake/resolve_dependency_modules/clp.cmake(1 hunks)
🧰 Additional context used
🧠 Learnings (1)
📓 Common learnings
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h:36-39
Timestamp: 2025-08-20T18:00:51.241Z
Learning: In ClpArchiveCursor.h, the clp_s::InputSource type is available through transitive inclusion via BaseClpCursor.h, which includes "clp_s/InputConfig.hpp" where InputSource is defined. The constructor parameter type is properly accessible without needing direct inclusion.
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/ClpVectorLoader.cpp:24-24
Timestamp: 2025-08-20T18:13:32.442Z
Learning: In ClpVectorLoader.cpp, the include of BaseClpCursor.h is necessary because the implementation uses ColumnType enum values (like ColumnType::Integer, ColumnType::Float) in a switch statement. While ClpVectorLoader.h has a forward declaration of ColumnType, the full definition from BaseClpCursor.h is required in the implementation file to access the enum values.
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: Build with GCC / Ubuntu release with resolve_dependency
🔇 Additional comments (1)
CMake/resolve_dependency_modules/clp.cmake (1)
19-19: Pin bumped to bfd4f60: confirm IR dependency & capture rationale
- bfd4f60 (2025-09-05) is the “fix(kv-ir): Fix compiler warnings found in #29” commit; delta from 19cd534 includes 11 commits adding webui, docker, kv-ir and presto-clp features plus fixes.
- Ensure this SHA includes the IR changes your code relies on, and in your commit message add a brief note or link to upstream #29 for traceability.
There was a problem hiding this comment.
Actionable comments posted: 9
♻️ Duplicate comments (16)
velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.cpp (3)
23-29: Defensive null check and suppress unused index.Guard against null filteredLogEvents_ and mark log_event_idx unused to avoid UB and warnings.
Apply:
auto ClpIrUnitHandler::handle_log_event( ::clp::ffi::KeyValuePairLogEvent log_event, size_t log_event_idx) -> ::clp::ffi::ir_stream::IRErrorCode { + (void)log_event_idx; + VELOX_CHECK_NOT_NULL(filteredLogEvents_); filteredLogEvents_->push_back( std::make_unique<::clp::ffi::KeyValuePairLogEvent>(std::move(log_event))); return ::clp::ffi::ir_stream::IRErrorCode::IRErrorCode_Success; }
31-37: Explicitly mark unused parameters and document limitation.Make intent clear and silence warnings.
auto ClpIrUnitHandler::handle_schema_tree_node_insertion( bool is_auto_generated, ::clp::ffi::SchemaTree::NodeLocator schema_tree_node_locator, std::shared_ptr<::clp::ffi::SchemaTree const> const& schema_tree) -> ::clp::ffi::ir_stream::IRErrorCode { + (void)is_auto_generated; + (void)schema_tree_node_locator; + (void)schema_tree; + // TODO: Support auto-generated keys ('@') in a future PR. return ::clp::ffi::ir_stream::IRErrorCode::IRErrorCode_Success; }
39-41: Optional: release capacity after large batches.Shrink capacity when appropriate to reduce memory footprint.
void ClpIrUnitHandler::clearFilteredLogEvents() { filteredLogEvents_->clear(); + filteredLogEvents_->shrink_to_fit(); }velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cpp (2)
28-37: Avoid map copy and use correct index type.Use vector_size_t, take the map by const reference, and replace count+at with find.
- for (int vectorIndex : rows) { + for (vector_size_t vectorIndex : rows) { auto& logEvent = filteredLogEvents_->at(vectorIndex); // TODO: also need to support auto-generated keys - auto userGenNodeIdValueMap = logEvent->get_user_gen_node_id_value_pairs(); + const auto& userGenNodeIdValueMap = + logEvent->get_user_gen_node_id_value_pairs(); vector->setNull(vectorIndex, true); - if (0 == userGenNodeIdValueMap.count(nodeId_)) { - continue; - } - auto value = userGenNodeIdValueMap.at(nodeId_); + const auto it = userGenNodeIdValueMap.find(nodeId_); + if (it == userGenNodeIdValueMap.end()) { + continue; + } + const auto& value = it->second;
90-126: Array loading corrupts offsets and uses dangling string_views.Offset is always 0, overwriting previous rows; string_views point to temporaries from to_json_string(). Fix by appending elements and setting per-row offset/size.
case ColumnType::Array: { auto arrayVector = std::dynamic_pointer_cast<ArrayVector>(vector); std::string jsonString; @@ - size_t numElements{0ULL}; - auto elements = arrayVector->elements()->asFlatVector<StringView>(); - auto obj = arrayParser_->iterate(jsonString); - std::vector<std::string_view> rawElements; - for (auto arrayElement : obj.get_array()) { - auto raw_element = simdjson::to_json_string(arrayElement).value(); - rawElements.emplace_back(raw_element); - } - elements->resize(rawElements.size()); - for (auto& raw_element : rawElements) { - elements->set(numElements++, StringView(raw_element)); - } - arrayVector->setOffsetAndSize(vectorIndex, 0ULL, numElements); + auto elements = arrayVector->elements()->asFlatVector<StringView>(); + auto doc = arrayParser_.iterate(jsonString); + vector_size_t start = elements->size(); + vector_size_t count = 0; + for (auto elem : doc.get_array()) { + auto sres = simdjson::to_json_string(elem); + if (sres.error()) { + continue; + } + const std::string& s = sres.value(); + elements->resize(start + count + 1); + elements->set(start + count, StringView(s)); + ++count; + } + arrayVector->setOffsetAndSize(vectorIndex, start, count); arrayVector->setNull(vectorIndex, false); break; }Note: This assumes arrayParser_ is a by-value thread_local (see header comment).
velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h (2)
21-25: Include VectorLoader explicitly to avoid brittle transitive includes.#include "connectors/clp/search_lib/BaseClpCursor.h" #include "ffi/ir_stream/Deserializer.hpp" +#include "velox/vector/VectorLoader.h" #include "velox/vector/FlatVector.h" #include "velox/vector/LazyVector.h"
41-42: Store simdjson parser by value; drop heap indirection.Cheaper init, fewer pointers, simpler lifetime. Update .cpp to call arrayParser_.iterate(...).
- inline static thread_local std::unique_ptr<simdjson::ondemand::parser> - arrayParser_ = std::make_unique<simdjson::ondemand::parser>(); + inline static thread_local simdjson::ondemand::parser arrayParser_{};velox/connectors/clp/search_lib/ir/ClpIrCursor.h (3)
19-22: Add required standard headers; avoid transitive include fragility.#pragma once +#include <functional> +#include <memory> +#include <string> +#include <string_view> +#include <unordered_map> +#include <utility> +#include <vector> + #include "ffi/ir_stream/Deserializer.hpp" #include "velox/connectors/clp/search_lib/BaseClpCursor.h" #include "velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h"
23-23: Forward-declare clp::ReaderInterface to satisfy pointer member.namespace facebook::velox::connector::clp::search_lib { +namespace clp { +class ReaderInterface; +} // namespace clp
77-79: Include the header that defines literal_type_bitmask_t.- std::vector< - std::pair<std::string, clp_s::search::ast::literal_type_bitmask_t>> + std::vector< + std::pair<std::string, clp_s::search::ast::literal_type_bitmask_t>> splitFieldsToNamesAndTypes() const;Add (near other includes at top of this header):
+#include "clp_s/search/ast/Types.hpp" // for literal_type_bitmask_tvelox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h (2)
19-23: Add missing standard include for size_t.Include to declare size_t explicitly.
#include <memory> #include <vector> +#include <cstddef> #include "ffi/ir_stream/Deserializer.hpp"
60-67: Avoid refcount churn and mark accessors noexcept.Return the shared_ptr by const-reference and mark both methods noexcept.
- std::shared_ptr< - const std::vector<std::unique_ptr<::clp::ffi::KeyValuePairLogEvent>>> - getFilteredLogEvents() const { - return filteredLogEvents_; - } + const std::shared_ptr< + const std::vector<std::unique_ptr<::clp::ffi::KeyValuePairLogEvent>>>>& + getFilteredLogEvents() const noexcept { + return filteredLogEvents_; + } @@ - void clearFilteredLogEvents(); + void clearFilteredLogEvents() noexcept;Also update the implementation:
// velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.cpp -void ClpIrUnitHandler::clearFilteredLogEvents() { +void ClpIrUnitHandler::clearFilteredLogEvents() noexcept { filteredLogEvents_->clear(); }velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp (4)
41-49: Handle deserialization failure without aborting.Avoid VELOX_FAIL; set errorCode_ and return 0 so the pipeline can recover.
if (deserializeResult.has_error()) { auto error = deserializeResult.error(); - VELOX_FAIL( - "IR file {} might be broken, failed to deserialize. {}: {}", - this->splitPath_, - error.category().name(), - error.message()); + VLOG(2) << "IR deserialize failed for \"" << splitPath_ << "\": " + << error.category().name() << ": " << error.message(); + errorCode_ = ErrorCode::InternalError; + return 0; }
50-51: Align fetchNext() return value with BaseClpCursor behaviour.Return the number of filtered rows, not deserialized events count.
- return irDeserializer_->get_num_log_events_deserialized(); + return getNumFilteredRows();
156-161: Compare std::error_code correctly to std::errc.Use make_error_code to ensure category matches; also covered by the new include.
- if (std::errc::result_out_of_range == error || + if (error == std::make_error_code(std::errc::result_out_of_range) || irDeserializer_->is_stream_completed()) { break; }
17-23: Include <system_error> for std::errc comparisons.Required to safely compare std::error_code to std::errc.
#include "ffi/ir_stream/search/QueryHandler.hpp" #include "velox/connectors/clp/search_lib/ir/ClpIrCursor.h" #include "velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h" +#include <system_error>
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (6)
velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp(1 hunks)velox/connectors/clp/search_lib/ir/ClpIrCursor.h(1 hunks)velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.cpp(1 hunks)velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h(1 hunks)velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cpp(1 hunks)velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h(1 hunks)
🧰 Additional context used
🧠 Learnings (3)
📓 Common learnings
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h:36-39
Timestamp: 2025-08-20T18:00:51.241Z
Learning: In ClpArchiveCursor.h, the clp_s::InputSource type is available through transitive inclusion via BaseClpCursor.h, which includes "clp_s/InputConfig.hpp" where InputSource is defined. The constructor parameter type is properly accessible without needing direct inclusion.
📚 Learning: 2025-08-20T18:13:32.442Z
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/ClpVectorLoader.cpp:24-24
Timestamp: 2025-08-20T18:13:32.442Z
Learning: In ClpVectorLoader.cpp, the include of BaseClpCursor.h is necessary because the implementation uses ColumnType enum values (like ColumnType::Integer, ColumnType::Float) in a switch statement. While ClpVectorLoader.h has a forward declaration of ColumnType, the full definition from BaseClpCursor.h is required in the implementation file to access the enum values.
Applied to files:
velox/connectors/clp/search_lib/ir/ClpIrCursor.hvelox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cppvelox/connectors/clp/search_lib/ir/ClpIrCursor.cppvelox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h
📚 Learning: 2025-08-20T18:00:51.241Z
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h:36-39
Timestamp: 2025-08-20T18:00:51.241Z
Learning: In ClpArchiveCursor.h, the clp_s::InputSource type is available through transitive inclusion via BaseClpCursor.h, which includes "clp_s/InputConfig.hpp" where InputSource is defined. The constructor parameter type is properly accessible without needing direct inclusion.
Applied to files:
velox/connectors/clp/search_lib/ir/ClpIrCursor.h
🧬 Code graph analysis (4)
velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h (3)
velox/connectors/clp/search_lib/ir/ClpIrCursor.h (1)
velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h (1)
velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.cpp (4)
handle_log_event(23-29)handle_log_event(23-25)clearFilteredLogEvents(39-41)clearFilteredLogEvents(39-39)
velox/connectors/clp/search_lib/ir/ClpIrCursor.h (4)
velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h (2)
ClpIrUnitHandler(26-46)velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h (1)
velox/connectors/clp/search_lib/BaseClpCursor.h (1)
BaseClpCursor(68-137)velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp (14)
fetchNext(28-51)fetchNext(28-28)getNumFilteredRows(53-55)getNumFilteredRows(53-53)createVector(57-68)createVector(57-60)loadSplit(70-109)loadSplit(70-70)splitFieldsToNamesAndTypes(111-146)splitFieldsToNamesAndTypes(112-112)deserialize(148-166)deserialize(148-149)createVectorHelper(168-206)createVectorHelper(168-171)
velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp (2)
velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h (1)
readerIndex_(64-64)velox/connectors/clp/search_lib/BaseClpCursor.h (1)
currentSplitLoaded_(122-135)
velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h (4)
velox/connectors/clp/search_lib/ir/ClpIrCursor.h (1)
velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h (1)
velox/vector/LazyVector.h (1)
VectorLoader(177-210)velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cpp (2)
loadInternal(22-131)loadInternal(22-26)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: Build with GCC / Ubuntu release with resolve_dependency
🔇 Additional comments (7)
velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.cpp (1)
19-19: Include path fix looks correct.Header path now matches Velox layout. No action required.
velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h (1)
30-38: Ctor takes shared_ptr by const-ref — good.Avoids an extra shared_ptr control-block bump. No change needed.
velox/connectors/clp/search_lib/ir/ClpIrCursor.h (1)
33-36: Non-copyable/movable — good mitigation for capturingthisin callback.Prevents dangling captures. No change needed.
velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h (1)
68-72: Document thread-safety/ownership of filteredLogEvents_.Clarify whether ClpIrUnitHandler is thread-confined and if callers may retain the returned shared_ptr across fetchNext() calls.
velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp (3)
97-106: Good: move ownership of handler and query handler.Passing irHandler and queryHandler by move avoids extra copies and lifetime ambiguity.
111-146: Timestamp mapping contradicts PR note—avoid misleading type.IR doesn’t populate Timestamp yet; mapping to EpochDateT risks mismatched projections. Map to UnknownT (or gate behind a capability flag).
- case ColumnType::Timestamp: - // TODO: IR timestamp support pending; constrain to Unknown to avoid - // mismatched projections. - literalType = search::ast::LiteralType::EpochDateT; + case ColumnType::Timestamp: + // IR timestamp support pending; avoid mismatched projections. + literalType = search::ast::LiteralType::UnknownT; break;
190-195: Good: defensive lookup with explicit check.Replacing at() with find()+VELOX_CHECK improves diagnostics.
| std::vector<VectorPtr> children; | ||
| auto& rowType = vectorType->as<TypeKind::ROW>(); | ||
| for (uint32_t i = 0; i < rowType.size(); ++i) { | ||
| children.push_back( | ||
| createVectorHelper(pool, rowType.childAt(i), vectorSize)); | ||
| } |
There was a problem hiding this comment.
🧹 Nitpick (assertive)
Pre-reserve children vector capacity.
Minor allocation win for ROW vectors.
- std::vector<VectorPtr> children;
+ std::vector<VectorPtr> children;
+ children.reserve(rowType.size());Committable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
In velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp around lines 173 to 178,
the children vector is built by repeatedly pushing back without reserving
capacity which causes extra allocations; call children.reserve(rowType.size())
before the loop (use appropriate integer type for reserve if needed) so the
vector pre-allocates space for all ROW child vectors and avoids repeated
reallocations.
| std::shared_ptr< | ||
| ::clp::ffi::ir_stream::Deserializer<ClpIrUnitHandler, QueryHandlerType>> | ||
| irDeserializer_; | ||
| std::shared_ptr<::clp::ReaderInterface> irReader_{nullptr}; | ||
| std::unordered_map<std::string, ::clp::ffi::SchemaTree::Node::id_t> | ||
| projectedColumnNameNodeIdMap_; | ||
| size_t readerIndex_{0}; |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Clear projection map and reset reader index when loading a new split.
Avoid stale mappings across splits.
Apply in ClpIrCursor.cpp (loadSplit()):
ErrorCode ClpIrCursor::loadSplit() {
+ projectedColumnNameNodeIdMap_.clear();
+ readerIndex_ = 0;
auto networkAuthOption = inputSource_ == InputSource::Filesystem
? NetworkAuthOption{.method = AuthMethod::None}
: NetworkAuthOption{.method = AuthMethod::S3PresignedUrlV4};📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| std::shared_ptr< | |
| ::clp::ffi::ir_stream::Deserializer<ClpIrUnitHandler, QueryHandlerType>> | |
| irDeserializer_; | |
| std::shared_ptr<::clp::ReaderInterface> irReader_{nullptr}; | |
| std::unordered_map<std::string, ::clp::ffi::SchemaTree::Node::id_t> | |
| projectedColumnNameNodeIdMap_; | |
| size_t readerIndex_{0}; | |
| ErrorCode ClpIrCursor::loadSplit() { | |
| projectedColumnNameNodeIdMap_.clear(); | |
| readerIndex_ = 0; | |
| auto networkAuthOption = inputSource_ == InputSource::Filesystem | |
| ? NetworkAuthOption{.method = AuthMethod::None} | |
| : NetworkAuthOption{.method = AuthMethod::S3PresignedUrlV4}; | |
| // ... rest of loadSplit implementation | |
| } |
🤖 Prompt for AI Agents
In velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp in the loadSplit()
implementation, the projection map projectedColumnNameNodeIdMap_ and
readerIndex_ are not reset when loading a new split, allowing stale mappings and
reader position to leak across splits; clear projectedColumnNameNodeIdMap_
(e.g., projectedColumnNameNodeIdMap_.clear()) and reset readerIndex_ to 0 at the
start of loadSplit() after any error/reader setup to ensure a fresh state for
each split.
| ClpIrUnitHandler() { | ||
| filteredLogEvents_ = std::make_shared< | ||
| std::vector<std::unique_ptr<::clp::ffi::KeyValuePairLogEvent>>>(); | ||
| } |
There was a problem hiding this comment.
🧹 Nitpick (assertive)
Prefer member-initializer list for shared_ptr construction.
Slightly cleaner and avoids a default-initialization followed by assignment.
- ClpIrUnitHandler() {
- filteredLogEvents_ = std::make_shared<
- std::vector<std::unique_ptr<::clp::ffi::KeyValuePairLogEvent>>>();
- }
+ ClpIrUnitHandler()
+ : filteredLogEvents_{std::make_shared<
+ std::vector<std::unique_ptr<::clp::ffi::KeyValuePairLogEvent>>>()} {}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| ClpIrUnitHandler() { | |
| filteredLogEvents_ = std::make_shared< | |
| std::vector<std::unique_ptr<::clp::ffi::KeyValuePairLogEvent>>>(); | |
| } | |
| ClpIrUnitHandler() | |
| : filteredLogEvents_{std::make_shared< | |
| std::vector<std::unique_ptr<::clp::ffi::KeyValuePairLogEvent>>>()} {} |
🤖 Prompt for AI Agents
In velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h around lines 28 to 31,
the constructor default-initializes filteredLogEvents_ and then assigns a
shared_ptr in the body; change this to initialize filteredLogEvents_ directly in
the member-initializer list using
std::make_shared<std::vector<std::unique_ptr<::clp::ffi::KeyValuePairLogEvent>>>(),
and remove the assignment from the constructor body so the member is constructed
in-place.
| [[nodiscard]] auto handle_log_event( | ||
| ::clp::ffi::KeyValuePairLogEvent log_event, | ||
| size_t log_event_idx) -> ::clp::ffi::ir_stream::IRErrorCode; |
There was a problem hiding this comment.
🧹 Nitpick (assertive)
Silence unused parameter warning.
Mark log_event_idx as maybe_unused.
[[nodiscard]] auto handle_log_event(
::clp::ffi::KeyValuePairLogEvent log_event,
- size_t log_event_idx) -> ::clp::ffi::ir_stream::IRErrorCode;
+ [[maybe_unused]] size_t log_event_idx) -> ::clp::ffi::ir_stream::IRErrorCode;📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| [[nodiscard]] auto handle_log_event( | |
| ::clp::ffi::KeyValuePairLogEvent log_event, | |
| size_t log_event_idx) -> ::clp::ffi::ir_stream::IRErrorCode; | |
| [[nodiscard]] auto handle_log_event( | |
| ::clp::ffi::KeyValuePairLogEvent log_event, | |
| [[maybe_unused]] size_t log_event_idx) -> ::clp::ffi::ir_stream::IRErrorCode; |
🤖 Prompt for AI Agents
In velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h around lines 37-39, the
parameter log_event_idx currently triggers an unused-parameter warning; mark it
as maybe_unused by annotating the parameter in the function declaration (e.g.,
add the [[maybe_unused]] attribute before the parameter name) so the compiler
knows it may be unused and the warning is suppressed.
| void ClpIrVectorLoader::loadInternal( | ||
| RowSet rows, | ||
| ValueHook* hook, | ||
| vector_size_t resultSize, | ||
| VectorPtr* result) { | ||
| auto vector = *result; |
There was a problem hiding this comment.
💡 Verification agent
🧩 Analysis chain
Respect VectorLoader’s ValueHook contract (or assert unsupported).
Current implementation ignores hook, which violates the documented contract and can hurt performance.
Would you like me to add hook handling for primitive types now (String/Integer/Float/Boolean) and assert unsupported for Array?
🏁 Script executed:
#!/usr/bin/env bash
# Search for VectorLoader definition to inspect its ValueHook contract
rg -nP "class\s+VectorLoader" -C3
# Locate the ValueHook declaration to confirm its intended usage
rg -nP "class\s+ValueHook" -C3
# Find all loadInternal implementations to compare how other loaders honour the hook
rg -nP "loadInternal\s*\(" -g "*.cpp" -C3Length of output: 5875
🏁 Script executed:
#!/usr/bin/env bash
# Inspect ValueHook contract in LazyVector.h
sed -n '1,120p' velox/vector/LazyVector.h
# Inspect ClpIrVectorLoader implementation to confirm hook is ignored
sed -n '1,50p' velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cppLength of output: 5585
Respect ValueHook contract in ClpIrVectorLoader
ClpIrVectorLoader::loadInternal ignores the hook parameter, violating the ValueHook contract in VectorLoader (loaded values must be forwarded to the hook to enable pushdown and avoid unnecessary vector writes). Either invoke hook->addValue/hook->addNull for each row when hook != nullptr, or assert unsupported (e.g. VELOX_UNSUPPORTED()) if hook-based loading isn’t implemented.
🤖 Prompt for AI Agents
In velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cpp around lines 22-27,
loadInternal currently ignores the ValueHook* hook which violates the
VectorLoader contract; update the implementation to respect the hook: if hook !=
nullptr, iterate the input RowSet and for each row call hook->addNull(rowIndex)
for nulls or hook->addValue(rowIndex, referenceOrSerializedValue) for non-nulls
(forward the value bytes/representation expected by the hook) instead of writing
directly into *result; if hook-based loading is not supported, explicitly fail
with VELOX_UNSUPPORTED() (or an equivalent assertion) rather than silently
ignoring the hook.
| break; | ||
| } | ||
| case ColumnType::Array: { | ||
| auto arrayVector = std::dynamic_pointer_cast<ArrayVector>(vector); |
There was a problem hiding this comment.
🧹 Nitpick (assertive)
Prefer as() over dynamic_pointer_cast for clarity.
Avoid shared_ptr RTTI; use vector->as().
- auto arrayVector = std::dynamic_pointer_cast<ArrayVector>(vector);
+ auto* arrayVector = vector->as<ArrayVector>();📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| auto arrayVector = std::dynamic_pointer_cast<ArrayVector>(vector); | |
| // Before: | |
| - auto arrayVector = std::dynamic_pointer_cast<ArrayVector>(vector); | |
| // After: | |
| auto* arrayVector = vector->as<ArrayVector>(); |
🤖 Prompt for AI Agents
In velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cpp around line 91,
replace the use of std::dynamic_pointer_cast<ArrayVector>(vector) with the
Velox-style RTTI helper by calling vector->as<ArrayVector>() to avoid shared_ptr
RTTI; update the assignment accordingly and ensure the variable type matches the
returned reference/pointer semantics (use auto& or auto depending on expected
ownership).
| private: | ||
| inline static thread_local std::unique_ptr<simdjson::ondemand::parser> | ||
| arrayParser_ = std::make_unique<simdjson::ondemand::parser>(); | ||
|
|
||
| ColumnType nodeType_; | ||
| ::clp::ffi::SchemaTree::Node::id_t nodeId_; | ||
| std::shared_ptr< | ||
| const std::vector<std::unique_ptr<::clp::ffi::KeyValuePairLogEvent>>> | ||
| filteredLogEvents_; | ||
|
|
There was a problem hiding this comment.
🧹 Nitpick (assertive)
Document thread-safety guarantees.
arrayParser_ is thread_local; clarify that ClpIrVectorLoader is thread-safe for concurrent loadInternal calls across threads but not re-entrant per instance.
gibber9809
left a comment
gibber9809
left a comment
There was a problem hiding this comment.
Partial review for code under the ir subdirectory. Still need to review ClpIrVectorLoader.cpp and ClpIrVectorLoader.h, but those two files should be more straightforward.
| clp-s-ir-search | ||
| PUBLIC clp_s::archive_reader | ||
| PRIVATE | ||
| clp_s::clp_dependencies |
There was a problem hiding this comment.
We should make a note to export library targets for the IR stream code inside of the CLP repo -- relying on clp_s::clp_dependencies will work for now, but isn't ideal long term.
There was a problem hiding this comment.
How about I adding a comment "# After IR-stream-related targets are exported, we should replace this with finer granularity dependencies" above clp_s::clp_dependencies
There was a problem hiding this comment.
# Once the IR-stream-related targets are exported, this should be updated to use more fine-grained dependencies.
| auto ClpIrUnitHandler::handle_log_event( | ||
| ::clp::ffi::KeyValuePairLogEvent log_event, | ||
| size_t log_event_idx) -> ::clp::ffi::ir_stream::IRErrorCode { | ||
| filteredLogEvents_->push_back( | ||
| std::make_unique<::clp::ffi::KeyValuePairLogEvent>(std::move(log_event))); | ||
| return ::clp::ffi::ir_stream::IRErrorCode::IRErrorCode_Success; | ||
| } | ||
|
|
||
| auto ClpIrUnitHandler::handle_schema_tree_node_insertion( | ||
| bool is_auto_generated, | ||
| ::clp::ffi::SchemaTree::NodeLocator schema_tree_node_locator, | ||
| std::shared_ptr<::clp::ffi::SchemaTree const> const& schema_tree) | ||
| -> ::clp::ffi::ir_stream::IRErrorCode { | ||
| return ::clp::ffi::ir_stream::IRErrorCode::IRErrorCode_Success; | ||
| } | ||
|
|
||
| void ClpIrUnitHandler::clearFilteredLogEvents() { | ||
| filteredLogEvents_->clear(); | ||
| } |
There was a problem hiding this comment.
These functions seem pretty short. Maybe we could just leave all the implementations in the header?
There was a problem hiding this comment.
It seems these are the only three functions. Shall we only keep the .h file after moving the implementation to the header?
| projected_key_and_index) | ||
| -> ystdlib::error_handling::Result<void> { | ||
| projectedColumnNameNodeIdMap_.insert( | ||
| {std::string(projected_key_and_index.first), nodeId}); |
There was a problem hiding this comment.
Now that this callback also gives the index of the resolved projection within the original list of projections I think we should use that instead of the name -- it will allow this code to work correctly for cases where there are duplicate columns in the projection, and when there are two columns with the same name but different projected type.
So basically change projectedColumnNameNodeIdMap to a std::unordered_map<size_t, ::clp::ffi::SchemaTree::Node::id_t and store the mapping between projected_key_and_index.second and nodeId. Then later when we're checking what each projected column maps to instead of going index -> name -> projected nodes we can just directly look up index -> projected nodes.
| "Reader index out of bounds"); | ||
| auto projectedColumn = outputColumns_[readerIndex_]; | ||
| auto projectedColumnType = projectedColumn.type; | ||
| auto it = projectedColumnNameNodeIdMap_.find(projectedColumn.name); |
There was a problem hiding this comment.
Related to my other comment -- we should end up being able to directly look up projectedColumnIndexNodeIdMap_.find(readerIndex_).
| vector->setNulls(allocateNulls(vectorSize, pool, bits::kNull)); | ||
| VELOX_CHECK_LT( | ||
| readerIndex_, | ||
| projectedColumnNameNodeIdMap_.size(), |
There was a problem hiding this comment.
| projectedColumnNameNodeIdMap_.size(), | |
| outputColumns_.size(), |
Otherwise we would crash in situations where not all columns have been resolved yet when we should just be marshalling nulls for unresolved fields.
| VELOX_CHECK( | ||
| it != projectedColumnNameNodeIdMap_.end(), | ||
| "Projected column '{}' not found in node id map", | ||
| projectedColumn.name); |
There was a problem hiding this comment.
Related to my other comment -- instead of asserting here we should be marshalling nulls for fields that haven't resolved to any specific nodes.
| } | ||
| return error; | ||
| } | ||
| cnt++; |
There was a problem hiding this comment.
I think we're trying to count matching results here right? In that case we need to rely on the fact that for search the handle_log_event method on the IrUnitHandler is only invoked for matching results. This current code will actually count for every single IR unit which will include non-matching log events, new schema tree nodes, etc..
I think we should be able to do something like whille(ir_unit_handler.get_num_filtered_log_events() < numRows) instead.
There was a problem hiding this comment.
Here actually we are counting the scanned log events. The numRows is the current batch size, the cnt here is to ensure it doesn't scan more than batch size. The number of matching results returns by getNumFilteredRows function above.
Do you think we change the filtered to matching? The filtered is inherited from previous code.
There was a problem hiding this comment.
I see, in that case what we should do is something like
if (clp::ffi::ir_stream::IrUnitType::LogEvent == deserializeResult.value()) {
++cnt;
}
to make sure we're just counting scanned log events instead of counting all IR units.
| if (deserializeResult.has_error()) { | ||
| auto error = deserializeResult.error(); | ||
| if (std::errc::result_out_of_range == error || | ||
| irDeserializer_->is_stream_completed()) { | ||
| break; | ||
| } | ||
| return error; |
There was a problem hiding this comment.
Judging by some comments in the clp code it looks like we might switch from result_out_of_range to another error code to indicate truncated IR streams in the future -- we should probably put up an issue or something to track that so that we remember to change this code at the same time.
There was a problem hiding this comment.
Yes. We were planning to add our own error code for IR serde. I think we will have it in at some point.
There was a problem hiding this comment.
Do we have an issue link so that I can put it in the code comment?
gibber9809
left a comment
gibber9809
left a comment
There was a problem hiding this comment.
Partial review for the ir subdirectory, covering the code I did not look at in my first partial review.
| inline static thread_local std::unique_ptr<simdjson::ondemand::parser> | ||
| arrayParser_ = std::make_unique<simdjson::ondemand::parser>(); |
There was a problem hiding this comment.
Does this need to be a unique_ptr? Also, instead of having this be a static thread_local member maybe we could just have it be a regular member variable in ClpIrCursor and pass it down for marshalling? Having this as a static thread local variable makes the code harder to reason about, especially when debugging errors.
There was a problem hiding this comment.
I guess since this follows the LazyVector model it might be a bit difficult to have it be a ClpIRCursor member. Even still I think the code is easier to interpret if either
- We just directly make this member variable
simdjson::ondemand::parser arrayParser_or - We wrap it in optional
std::optional<simdjson::ondemand::parser>and initialize it only when needed if we're that concerned about the performance overhead of initializing/allocating buffers for the parser.
There was a problem hiding this comment.
Nice catch. This was copied from the ClpArchiveVectorLoader, didn't notice this previously.
| literalType = search::ast::LiteralType::IntegerT; | ||
| break; | ||
| case ColumnType::String: | ||
| literalType = search::ast::LiteralType::VarStringT; |
There was a problem hiding this comment.
| literalType = search::ast::LiteralType::VarStringT; | |
| literalType = search::ast::LiteralType::VarStringT | search::ast::LiteralType::ClpStringT; |
| if (0 == userGenNodeIdValueMap.count(nodeId_)) { | ||
| continue; | ||
| } | ||
| auto value = userGenNodeIdValueMap.at(nodeId_); | ||
| if (!value.has_value()) { | ||
| continue; | ||
| } |
There was a problem hiding this comment.
| if (0 == userGenNodeIdValueMap.count(nodeId_)) { | |
| continue; | |
| } | |
| auto value = userGenNodeIdValueMap.at(nodeId_); | |
| if (!value.has_value()) { | |
| continue; | |
| } | |
| auto const value_it{userGenNodeIdValueMap.find(nodeId_)}; | |
| if (userGenNodeIdValueMap.end() == value_it || false == value_it->has_value()) { | |
| continue; | |
| } | |
| auto const& value{*value_it}; |
Maybe something more like this? Avoids doing the lookup twice + makes the special case handling a bit easier to follow imo.
e652870 to
ef62d9b
Compare
|
Addressed the comments of the first round review. Added the IR stream file for the |
There was a problem hiding this comment.
Actionable comments posted: 6
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
velox/connectors/clp/tests/ClpConnectorTest.cpp (1)
83-87: Prefer path joins over string concatenation for portability.Apply:
static std::string getExampleFilePath(const std::string& filePath) { - std::string current_path = fs::current_path().string(); - return current_path + "/examples/" + filePath; + const auto base = fs::current_path(); + return (base / "examples" / filePath).string(); }
♻️ Duplicate comments (24)
velox/connectors/clp/search_lib/ir/CMakeLists.txt (1)
23-33: Link against the IR stream library, not the archive reader.This target implements IR; linking PUBLIC to clp_s::archive_reader is incorrect and can cause unresolved IR symbols or accidental dependency coupling.
Apply:
velox_link_libraries( clp-s-ir-search - PUBLIC clp_s::archive_reader + PUBLIC clp_s::ir_stream # or your CLP export name for IR (e.g., clp_s::ir_reader) PRIVATE # Once the IR-stream-related targets are exported, this should be updated to # use more fine-grained dependencies. clp_s::clp_dependencies clp_s::io clp_s::search clp_s::search::kql velox_vector)If the export name differs in your CLP version, substitute accordingly.
velox/connectors/clp/tests/ClpConnectorTest.cpp (1)
75-81: Stabilise order-sensitive assertions via single-driver execution.Apply:
RowVectorPtr getResults( const core::PlanNodePtr& planNode, std::vector<exec::Split>&& splits) { return exec::test::AssertQueryBuilder(planNode) + .maxDrivers(1) // ensure deterministic ordering .splits(std::move(splits)) .copyResults(pool()); }velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cpp (4)
22-27: Respect ValueHook contract or explicitly fail when provided.Apply:
void ClpIrVectorLoader::loadInternal( RowSet rows, ValueHook* hook, vector_size_t resultSize, VectorPtr* result) { + if (hook != nullptr) { + VELOX_UNSUPPORTED("ClpIrVectorLoader does not support ValueHook yet."); + } auto vector = *result;
93-93: Use Velox RTTI helper for vectors.Apply:
- auto arrayVector = std::dynamic_pointer_cast<ArrayVector>(vector); + auto* arrayVector = vector->as<ArrayVector>();
28-37: Avoid map copy; use correct index type.Apply:
- for (int vectorIndex : rows) { + for (vector_size_t vectorIndex : rows) { vector->setNull(vectorIndex, true); if (!isResolved_) { continue; } auto& logEvent = filteredLogEvents_->at(vectorIndex); // TODO: also need to support auto-generated keys - auto userGenNodeIdValueMap = logEvent->get_user_gen_node_id_value_pairs(); - auto const value_it{userGenNodeIdValueMap.find(nodeId_)}; - if (userGenNodeIdValueMap.end() == value_it || + const auto& userGenNodeIdValueMap = + logEvent->get_user_gen_node_id_value_pairs(); + const auto value_it = userGenNodeIdValueMap.find(nodeId_); + if (userGenNodeIdValueMap.end() == value_it || false == value_it->second.has_value()) { continue; }
92-127: Array loading corrupts offsets and risks dangling data; fix offset accounting and write elements inline.Apply:
- case ColumnType::Array: { - auto arrayVector = std::dynamic_pointer_cast<ArrayVector>(vector); + case ColumnType::Array: { + auto* arrayVector = vector->as<ArrayVector>(); std::string jsonString; if (value->is<::clp::ir::EightByteEncodedTextAst>()) { auto decodeResult = value->get_immutable_view<::clp::ir::EightByteEncodedTextAst>() .decode_and_unparse(); if (!decodeResult.has_value()) { continue; } jsonString = std::move(decodeResult.value()); } else { auto decodeResult = value->get_immutable_view<::clp::ir::FourByteEncodedTextAst>() .decode_and_unparse(); if (!decodeResult.has_value()) { continue; } jsonString = std::move(decodeResult.value()); } - size_t numElements{0ULL}; - auto elements = arrayVector->elements()->asFlatVector<StringView>(); - auto obj = arrayParser_.iterate(jsonString); - std::vector<std::string_view> rawElements; - for (auto arrayElement : obj.get_array()) { - auto raw_element = simdjson::to_json_string(arrayElement).value(); - rawElements.emplace_back(raw_element); - } - elements->resize(rawElements.size()); - for (auto& raw_element : rawElements) { - elements->set(numElements++, StringView(raw_element)); - } - arrayVector->setOffsetAndSize(vectorIndex, 0ULL, numElements); + auto elements = arrayVector->elements()->asFlatVector<StringView>(); + auto doc = arrayParser_.iterate(jsonString); + vector_size_t start = elements->size(); + vector_size_t count = 0; + for (auto elem : doc.get_array()) { + auto s_res = simdjson::to_json_string(elem); + if (s_res.error()) { + continue; + } + std::string s = std::move(s_res.value()); // own bytes before set() + elements->resize(start + count + 1); + elements->set(start + count, StringView(s)); + ++count; + } + arrayVector->setOffsetAndSize(vectorIndex, start, count); arrayVector->setNull(vectorIndex, false); break; }Notes:
- Previous code hard-coded offset 0 for every row and overwrote prior rows via elements->resize(...).
- Eliminates std::string_view to temporaries by setting elements immediately.
velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h (1)
21-25: Include VectorLoader.h directly to avoid brittle transitive include.Inheritence from VectorLoader currently relies on LazyVector.h. Include VectorLoader.h explicitly.
#include "ffi/ir_stream/Deserializer.hpp" -#include "velox/vector/FlatVector.h" -#include "velox/vector/LazyVector.h" +#include "velox/vector/VectorLoader.h" +#include "velox/vector/FlatVector.h" +#include "velox/vector/LazyVector.h"velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp (7)
24-24: Remove global using-directive.Avoid namespace pollution; qualify symbols or use narrow using-declarations.
-using namespace clp_s; +// Prefer explicit clp_s:: qualifiers or targeted using-declarations.
28-51: fetchNext return value violates BaseClpCursor semantics.Returns “deserialized events” instead of “rows scanned” (archive cursor returns scanned). Align behaviour or update the contract everywhere. Minimal: return getNumFilteredRows().
- return irDeserializer_->get_num_log_events_deserialized(); + return getNumFilteredRows();
41-49: Do not abort on deserialize error; surface a recoverable error.Replace VELOX_FAIL with non-fatal handling to avoid crashing pipelines.
- if (deserializeResult.has_error()) { - auto error = deserializeResult.error(); - VELOX_FAIL( - "IR file {} might be broken, failed to deserialize. {}: {}", - this->splitPath_, - error.category().name(), - error.message()); - } + if (deserializeResult.has_error()) { + const auto error = deserializeResult.error(); + VLOG(2) << "IR deserialize failed for \"" << splitPath_ << "\": " + << error.category().name() << ": " << error.message(); + errorCode_ = ErrorCode::InternalError; + return 0; + }
156-160: Compare std::error_code against std::errc correctly.Use make_error_code for category-safe comparison; include <system_error>.
- if (std::errc::result_out_of_range == error || + if (error == std::make_error_code(std::errc::result_out_of_range) || irDeserializer_->is_stream_completed()) { break; }Also add:
+#include <system_error>
173-179: Pre-reserve children capacity for ROW vectors.Minor allocation win in deep schemas.
if (vectorType->kind() == TypeKind::ROW) { - std::vector<VectorPtr> children; + std::vector<VectorPtr> children; + children.reserve(vectorType->as<TypeKind::ROW>().size());
61-67: Equality check is too strict; unresolved projections are valid.projectedColumnIdNodeIdMap_.size() can be < outputColumns_.size() while unresolved columns should marshal nulls. Use <= or drop the check.
- VELOX_CHECK_EQ( - projectedColumnIdNodeIdMap_.size(), - outputColumns_.size(), - "Projected columns size {} does not match fields size {}", - projectedColumnIdNodeIdMap_.size(), - outputColumns_.size()); + VELOX_CHECK_LE( + projectedColumnIdNodeIdMap_.size(), + outputColumns_.size(), + "Projected columns map (size {}) cannot exceed projected fields (size {}).", + projectedColumnIdNodeIdMap_.size(), + outputColumns_.size());
135-139: Timestamp mapping contradicts the stated IR limitation.Comment says constrain to Unknown, but code maps to EpochDateT. Until IR timestamps are populated, avoid misleading type.
- case ColumnType::Timestamp: - // TODO: IR timestamp support pending; constrain to Unknown to avoid - // mismatched projections. - literalType = search::ast::LiteralType::EpochDateT; - break; + case ColumnType::Timestamp: + // TODO: IR timestamp support pending; avoid mismatched projections. + literalType = search::ast::LiteralType::UnknownT; + break;velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h (7)
19-26: Include missing standard headers.This header uses size_t but doesn’t include .
#include <memory> #include <vector> +#include <cstddef>
31-34: Prefer member-initialiser list for filteredLogEvents_.Cleaner and avoids re-assignment.
- ClpIrUnitHandler() { - filteredLogEvents_ = std::make_shared< - std::vector<std::unique_ptr<::clp::ffi::KeyValuePairLogEvent>>>(); - } + ClpIrUnitHandler() + : filteredLogEvents_{std::make_shared< + std::vector<std::unique_ptr<::clp::ffi::KeyValuePairLogEvent>>>()} {}
41-43: Silence unused parameter warning.Mark log_event_idx as maybe_unused.
- size_t log_event_idx) -> ::clp::ffi::ir_stream::IRErrorCode { + [[maybe_unused]] size_t log_event_idx) -> ::clp::ffi::ir_stream::IRErrorCode {
49-54: Tighten exception contract on trivial handlers.These do not throw; mark noexcept.
- [[nodiscard]] auto handle_utc_offset_change( + [[nodiscard]] auto handle_utc_offset_change( [[maybe_unused]] ::clp::UtcOffset utc_offset_old, [[maybe_unused]] ::clp::UtcOffset utc_offset_new) - -> ::clp::ffi::ir_stream::IRErrorCode { + noexcept -> ::clp::ffi::ir_stream::IRErrorCode {
65-67: Mark handle_end_of_stream noexcept.- [[nodiscard]] auto handle_end_of_stream() - -> ::clp::ffi::ir_stream::IRErrorCode { + [[nodiscard]] auto handle_end_of_stream() noexcept + -> ::clp::ffi::ir_stream::IRErrorCode {
70-74: Accessor should be noexcept.Trivial getter; mark noexcept.
- getFilteredLogEvents() const { + getFilteredLogEvents() const noexcept { return filteredLogEvents_; }
76-78: clearFilteredLogEvents should be noexcept.- void clearFilteredLogEvents() { + void clearFilteredLogEvents() noexcept { filteredLogEvents_->clear(); }velox/connectors/clp/search_lib/ir/ClpIrCursor.h (3)
19-22: Add missing standard and project includes; make the header self-contained.This header uses std::function, std::shared_ptr, std::vector, std::unordered_map, std::pair, std::string, std::string_view, size_t/uint64_t, and ystdlib::error_handling::Result, but doesn’t include their defining headers. Also include the type alias for literal_type_bitmask_t to avoid fragile transitive deps.
Apply:
#pragma once +#include <cstddef> +#include <cstdint> +#include <functional> +#include <memory> +#include <string> +#include <string_view> +#include <unordered_map> +#include <utility> +#include <vector> + +#include "clp_s/search/ast/Types.hpp" // literal_type_bitmask_t +#include "ystdlib/error_handling/Result.hpp" // Result/success + #include "ffi/ir_stream/Deserializer.hpp" #include "velox/connectors/clp/search_lib/BaseClpCursor.h" #include "velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h"If the exact paths differ in your repo, I can adjust them.
72-72: Forward-declare ::clp::ReaderInterface to satisfy the shared_ptr member.Avoid depending on a heavy header here; forward-declare the type used only via pointer.
#include "velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h" +namespace clp { +class ReaderInterface; +} // namespace clp + namespace facebook::velox::connector::clp::search_lib {
73-75: Reset projection map and reader index when loading a new split.Prevent stale mappings from previous splits and keep createVector() indexing correct.
Apply in ClpIrCursor.cpp (loadSplit()):
ErrorCode ClpIrCursor::loadSplit() { + projectedColumnIdNodeIdMap_.clear(); + readerIndex_ = 0; auto networkAuthOption = inputSource_ == InputSource::Filesystem ? NetworkAuthOption{.method = AuthMethod::None} : NetworkAuthOption{.method = AuthMethod::S3PresignedUrlV4};
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
📥 Commits
Reviewing files that changed from the base of the PR and between f9e1409 and ef62d9be4891ef5b64afd4d02bc86861593d78dd.
📒 Files selected for processing (7)
velox/connectors/clp/search_lib/ir/CMakeLists.txt(1 hunks)velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp(1 hunks)velox/connectors/clp/search_lib/ir/ClpIrCursor.h(1 hunks)velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h(1 hunks)velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cpp(1 hunks)velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h(1 hunks)velox/connectors/clp/tests/ClpConnectorTest.cpp(4 hunks)
🧰 Additional context used
🧠 Learnings (3)
📓 Common learnings
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h:36-39
Timestamp: 2025-08-20T18:00:51.241Z
Learning: In ClpArchiveCursor.h, the clp_s::InputSource type is available through transitive inclusion via BaseClpCursor.h, which includes "clp_s/InputConfig.hpp" where InputSource is defined. The constructor parameter type is properly accessible without needing direct inclusion.
📚 Learning: 2025-08-20T18:13:32.442Z
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/ClpVectorLoader.cpp:24-24
Timestamp: 2025-08-20T18:13:32.442Z
Learning: In ClpVectorLoader.cpp, the include of BaseClpCursor.h is necessary because the implementation uses ColumnType enum values (like ColumnType::Integer, ColumnType::Float) in a switch statement. While ClpVectorLoader.h has a forward declaration of ColumnType, the full definition from BaseClpCursor.h is required in the implementation file to access the enum values.
Applied to files:
velox/connectors/clp/search_lib/ir/ClpIrCursor.cppvelox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cppvelox/connectors/clp/search_lib/ir/ClpIrCursor.hvelox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h
📚 Learning: 2025-08-20T18:00:51.241Z
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h:36-39
Timestamp: 2025-08-20T18:00:51.241Z
Learning: In ClpArchiveCursor.h, the clp_s::InputSource type is available through transitive inclusion via BaseClpCursor.h, which includes "clp_s/InputConfig.hpp" where InputSource is defined. The constructor parameter type is properly accessible without needing direct inclusion.
Applied to files:
velox/connectors/clp/search_lib/ir/ClpIrCursor.h
🧬 Code graph analysis (4)
velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp (2)
velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h (1)
readerIndex_(64-64)velox/connectors/clp/search_lib/BaseClpCursor.h (1)
currentSplitLoaded_(122-135)
velox/connectors/clp/search_lib/ir/ClpIrCursor.h (4)
velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h (2)
ClpIrUnitHandler(29-47)velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h (1)
velox/connectors/clp/search_lib/BaseClpCursor.h (1)
BaseClpCursor(68-137)velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp (14)
fetchNext(28-51)fetchNext(28-28)getNumFilteredRows(53-55)getNumFilteredRows(53-53)createVector(57-68)createVector(57-60)loadSplit(70-109)loadSplit(70-70)splitFieldsToNamesAndTypes(111-147)splitFieldsToNamesAndTypes(112-112)deserialize(149-167)deserialize(149-150)createVectorHelper(169-206)createVectorHelper(169-172)
velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h (2)
velox/connectors/clp/search_lib/ir/ClpIrCursor.h (1)
velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h (1)
velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h (3)
velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h (1)
velox/vector/LazyVector.h (1)
VectorLoader(177-210)velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cpp (2)
loadInternal(22-133)loadInternal(22-26)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: Build with GCC / Ubuntu release with resolve_dependency
🔇 Additional comments (4)
velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cpp (1)
77-84: Align IR loader float type with ColumnType::Float semantics
IR loader (velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cpp:77–84) uses FlatVector/value_float_t, but archive loader (velox/connectors/clp/search_lib/archive/ClpArchiveVectorLoader.cpp:216–219) uses FlatVector. Confirm whether ColumnType::Float is 32-bit or 64-bit and, if it represents double-precision, update the IR loader to asFlatVector() and use value_double_t.velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h (1)
42-50: Array marshalling in the implementation has correctness risks (offset/size and dangling views).Based on ClpIrVectorLoader.cpp, two issues:
- elements->resize(rawElements.size()) resets the child size per row; offsets for prior rows will break.
- Storing std::string_view from to_json_string(...).value() risks dangling views; remove the intermediate storage and set directly while growing the child.
Proposed fix in ClpIrVectorLoader.cpp (illustrative patch):
- size_t numElements{0ULL}; - auto elements = arrayVector->elements()->asFlatVector<StringView>(); - auto obj = arrayParser_.iterate(jsonString); - std::vector<std::string_view> rawElements; - for (auto arrayElement : obj.get_array()) { - auto raw_element = simdjson::to_json_string(arrayElement).value(); - rawElements.emplace_back(raw_element); - } - elements->resize(rawElements.size()); - for (auto& raw_element : rawElements) { - elements->set(numElements++, StringView(raw_element)); - } - arrayVector->setOffsetAndSize(vectorIndex, 0ULL, numElements); + auto elements = arrayVector->elements()->asFlatVector<StringView>(); + auto doc = arrayParser_.iterate(jsonString); + auto arr = doc.get_array(); + const auto offset = elements->size(); + size_t length = 0; + for (auto v : arr) { + auto s = simdjson::to_json_string(v); + if (s.error()) { + continue; // or handle error + } + elements->resize(offset + length + 1); + elements->set(offset + length, StringView(s.value())); + ++length; + } + arrayVector->setOffsetAndSize(vectorIndex, offset, length); arrayVector->setNull(vectorIndex, false);Likely an incorrect or invalid review comment.
velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp (1)
149-166: Verify row limiting semantics match vector sizes.Loop counts IR units, not filtered log events; a unit may yield >1 event. Ensure getNumFilteredRows() never exceeds vectorSize. If needed, stop when filtered size reaches numRows.
Would you like a patch that bounds deserialisation by filtered count instead of unit count?
velox/connectors/clp/search_lib/ir/ClpIrCursor.h (1)
33-36: Good: non-copyable/movable + index-based projection mapping.Deleting copy/move prevents dangling captures; mapping by index resolves duplicate-name projections.
Also applies to: 73-74
There was a problem hiding this comment.
Use of uninitialised projectedColumnNodeId when unresolved.
projectedColumnNodeId is only set when isResolved is true, but passed unconditionally to ClpIrVectorLoader.
- ::clp::ffi::SchemaTree::Node::id_t projectedColumnNodeId;
- if (isResolved) {
- projectedColumnNodeId = it->second;
- }
+ ::clp::ffi::SchemaTree::Node::id_t projectedColumnNodeId{};
+ if (isResolved) {
+ projectedColumnNodeId = it->second;
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| ::clp::ffi::SchemaTree::Node::id_t projectedColumnNodeId; | |
| if (isResolved) { | |
| projectedColumnNodeId = it->second; | |
| } | |
| readerIndex_++; | |
| return std::make_shared<LazyVector>( | |
| pool, | |
| vectorType, | |
| vectorSize, | |
| std::make_unique<ClpIrVectorLoader>( | |
| isResolved, | |
| projectedColumnType, | |
| projectedColumnNodeId, | |
| irDeserializer_->get_ir_unit_handler().getFilteredLogEvents()), | |
| std::move(vector)); | |
| ::clp::ffi::SchemaTree::Node::id_t projectedColumnNodeId{}; | |
| if (isResolved) { | |
| projectedColumnNodeId = it->second; | |
| } | |
| readerIndex_++; | |
| return std::make_shared<LazyVector>( | |
| pool, | |
| vectorType, | |
| vectorSize, | |
| std::make_unique<ClpIrVectorLoader>( | |
| isResolved, | |
| projectedColumnType, | |
| projectedColumnNodeId, | |
| irDeserializer_->get_ir_unit_handler().getFilteredLogEvents()), | |
| std::move(vector)); |
🤖 Prompt for AI Agents
In velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp around lines 191 to 205,
projectedColumnNodeId is only assigned when isResolved is true but is
unconditionally passed to ClpIrVectorLoader; initialize projectedColumnNodeId at
declaration to a safe default (e.g. ::clp::ffi::SchemaTree::Node::id_t{} or an
explicit INVALID sentinel) or otherwise ensure it’s set for the unresolved path,
so the loader always receives a well-defined value (alternatively, modify
ClpIrVectorLoader to accept an optional/flag indicating unresolved state and
pass that instead).
There was a problem hiding this comment.
💡 Verification agent
🧩 Analysis chain
Unify type qualifier for literal_type_bitmask_t in decl/def and add the direct include.
Header uses clp_s::search::ast::literal_type_bitmask_t, but the .cpp shows search::ast::literal_type_bitmask_t. Make them consistent (prefer fully-qualified clp_s::search::ast::...).
Proposed .cpp fix:
-std::vector<std::pair<std::string, search::ast::literal_type_bitmask_t>>
+std::vector<std::pair<std::string, clp_s::search::ast::literal_type_bitmask_t>>
ClpIrCursor::splitFieldsToNamesAndTypes() const {Verification script:
🏁 Script executed:
#!/bin/bash
# Check for inconsistent qualifiers between header and source.
rg -n -C2 'literal_type_bitmask_t' velox/connectors/clp/search_lib/ir/ClpIrCursor.{h,cpp}
# Check if an alias like `using search = clp_s::search` exists.
rg -n -C2 -P 'using\s+search\s*=\s*clp_s::search' velox/connectors/clp/search_lib/ir/ClpIrCursor.cppLength of output: 1422
Unify literal_type_bitmask_t qualifier and add its direct include
- Header declares splitFieldsToNamesAndTypes() returning std::vector<std::pair<std::string, clp_s::search::ast::literal_type_bitmask_t>>, but the .cpp defines and uses search::ast::literal_type_bitmask_t. Update the .cpp signature and all occurrences to clp_s::search::ast::literal_type_bitmask_t.
- Include the header that defines literal_type_bitmask_t directly in ClpIrCursor.cpp.
🤖 Prompt for AI Agents
In velox/connectors/clp/search_lib/ir/ClpIrCursor.h around lines 77 to 79, the
header declares splitFieldsToNamesAndTypes() returning
std::vector<std::pair<std::string, clp_s::search::ast::literal_type_bitmask_t>>
but the implementation and usages in ClpIrCursor.cpp still reference
search::ast::literal_type_bitmask_t; update the ClpIrCursor.cpp function
signature and all occurrences to use the fully-qualified
clp_s::search::ast::literal_type_bitmask_t to match the header, and add an
#include for the header that defines literal_type_bitmask_t at the top of
ClpIrCursor.cpp so the type is resolved directly.
There was a problem hiding this comment.
🧹 Nitpick (assertive)
Narrow simdjson include to ondemand to reduce compile time.
Prefer the specific header for the used API.
-#include <simdjson.h>
+#include <simdjson/ondemand.h>📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| #include <simdjson.h> | |
| #include <simdjson/ondemand.h> |
🤖 Prompt for AI Agents
In velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h around lines 19 to 20,
the file currently includes the broad header <simdjson.h>; narrow this to the
specific ondemand API by replacing it with the ondemand header (e.g., include
<simdjson/ondemand.h>) so compilation pulls only the needed symbols and reduces
build time; update any symbol references if necessary to reflect the ondemand
namespace/header.
There was a problem hiding this comment.
🧹 Nitpick (assertive)
Document non-reentrancy/thread-safety of the loader.
arrayParser_ is an instance member; VectorLoader::load is called at most once, but clarify that the loader is not thread-safe nor re-entrant.
🤖 Prompt for AI Agents
In velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h around lines 28 to 40,
the class declaration/constructor omits documentation that this loader is not
thread-safe or re-entrant (arrayParser_ is an instance member and
VectorLoader::load is called at most once). Add a concise doc comment on the
class (or immediately above the constructor) stating that instances are not
thread-safe and must not be used concurrently or re-entered, mention that
arrayParser_ is stateful and VectorLoader::load is intended to be called at most
once per instance, and note expected ownership/usage constraints so callers know
to synchronize or create separate instances for concurrent use.
ef62d9b to
3962325
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (22)
velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h (2)
19-25: Include VectorLoader directly and narrow simdjson includeAvoid relying on transitive includes and pull only the ondemand API.
-#include <simdjson.h> +#include <simdjson/ondemand.h> #include "connectors/clp/search_lib/BaseClpCursor.h" #include "ffi/ir_stream/Deserializer.hpp" -#include "velox/vector/FlatVector.h" -#include "velox/vector/LazyVector.h" +#include "velox/vector/VectorLoader.h"
28-40: Document thread-safety/non-reentrancy of the loaderPlease add a brief class-level note: instances are not thread-safe or re-entrant; VectorLoader::loadInternal is called at most once per instance; arrayParser_ is stateful.
velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h (4)
31-34: Prefer member-initializer for shared_ptr constructionCleaner and avoids default-then-assign.
- ClpIrUnitHandler() { - filteredLogEvents_ = std::make_shared< - std::vector<std::unique_ptr<::clp::ffi::KeyValuePairLogEvent>>>(); - } + ClpIrUnitHandler() + : filteredLogEvents_{std::make_shared< + std::vector<std::unique_ptr<::clp::ffi::KeyValuePairLogEvent>>>()} {}
41-43: Silence unused parameter warningAnnotate log_event_idx as maybe_unused.
- ::clp::ffi::KeyValuePairLogEvent log_event, - size_t log_event_idx) -> ::clp::ffi::ir_stream::IRErrorCode { + ::clp::ffi::KeyValuePairLogEvent log_event, + [[maybe_unused]] size_t log_event_idx) -> ::clp::ffi::ir_stream::IRErrorCode {
49-54: Tighten exception contract on trivial handlersThese do not throw; mark them noexcept.
[[nodiscard]] auto handle_utc_offset_change( [[maybe_unused]] ::clp::UtcOffset utc_offset_old, [[maybe_unused]] ::clp::UtcOffset utc_offset_new) - -> ::clp::ffi::ir_stream::IRErrorCode { + noexcept -> ::clp::ffi::ir_stream::IRErrorCode { return ::clp::ffi::ir_stream::IRErrorCode::IRErrorCode_Success; } @@ - [[nodiscard]] auto handle_end_of_stream() - -> ::clp::ffi::ir_stream::IRErrorCode { + [[nodiscard]] auto handle_end_of_stream() noexcept + -> ::clp::ffi::ir_stream::IRErrorCode { return ::clp::ffi::ir_stream::IRErrorCode::IRErrorCode_Success; }Also applies to: 65-68
70-78: Getters/clear: noexcept and avoid refcount churnReturn the shared_ptr by const-reference and mark methods noexcept.
- std::shared_ptr< - const std::vector<std::unique_ptr<::clp::ffi::KeyValuePairLogEvent>>> - getFilteredLogEvents() const { - return filteredLogEvents_; - } + const std::shared_ptr< + const std::vector<std::unique_ptr<::clp::ffi::KeyValuePairLogEvent>>>& + getFilteredLogEvents() const noexcept { + return filteredLogEvents_; + } @@ - void clearFilteredLogEvents() { + void clearFilteredLogEvents() noexcept { filteredLogEvents_->clear(); }velox/connectors/clp/search_lib/ir/CMakeLists.txt (1)
23-33: Link against the IR stream library (not archive reader)This target uses ::clp::ffi::ir_stream symbols; link to the IR stream export.
velox_link_libraries( clp-s-ir-search - PUBLIC clp_s::archive_reader + PUBLIC clp_s::ir_stream PRIVATE # Once the IR-stream-related targets are exported, this should be updated to # use more fine-grained dependencies. clp_s::clp_dependencies clp_s::io clp_s::search clp_s::search::kql velox_vector)If your export name differs (e.g., clp_s::ir or clp_s::ir_reader), use that instead.
velox/connectors/clp/tests/ClpConnectorTest.cpp (1)
75-81: Stabilize order-sensitive assertionsForce single-driver execution in the helper to keep deterministic row order.
RowVectorPtr getResults( const core::PlanNodePtr& planNode, std::vector<exec::Split>&& splits) { - return exec::test::AssertQueryBuilder(planNode) + return exec::test::AssertQueryBuilder(planNode) + .maxDrivers(1) .splits(std::move(splits)) .copyResults(pool()); }velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cpp (3)
22-27: Respect ValueHook contract (or explicitly fail)Ignoring hook degrades pushdown and violates VectorLoader’s contract. Until supported, fail fast.
void ClpIrVectorLoader::loadInternal( RowSet rows, ValueHook* hook, vector_size_t resultSize, VectorPtr* result) { auto vector = *result; + if (hook != nullptr) { + VELOX_UNSUPPORTED("ValueHook is not supported by ClpIrVectorLoader yet."); + }
93-127: Array loading corrupts data: wrong offsets and dangling viewsEvery row points to offset 0 and intermediate string_views dangle. Append to elements, track start offset, and set immediately.
- case ColumnType::Array: { - auto arrayVector = std::dynamic_pointer_cast<ArrayVector>(vector); + case ColumnType::Array: { + auto* arrayVector = vector->as<ArrayVector>(); std::string jsonString; if (value->is<::clp::ir::EightByteEncodedTextAst>()) { auto decodeResult = value->get_immutable_view<::clp::ir::EightByteEncodedTextAst>() .decode_and_unparse(); if (!decodeResult.has_value()) { continue; } jsonString = std::move(decodeResult.value()); } else { auto decodeResult = value->get_immutable_view<::clp::ir::FourByteEncodedTextAst>() .decode_and_unparse(); if (!decodeResult.has_value()) { continue; } jsonString = std::move(decodeResult.value()); } - - size_t numElements{0ULL}; - auto elements = arrayVector->elements()->asFlatVector<StringView>(); - auto obj = arrayParser_.iterate(jsonString); - std::vector<std::string_view> rawElements; - for (auto arrayElement : obj.get_array()) { - auto raw_element = simdjson::to_json_string(arrayElement).value(); - rawElements.emplace_back(raw_element); - } - elements->resize(rawElements.size()); - for (auto& raw_element : rawElements) { - elements->set(numElements++, StringView(raw_element)); - } - arrayVector->setOffsetAndSize(vectorIndex, 0ULL, numElements); - arrayVector->setNull(vectorIndex, false); + auto* elements = arrayVector->elements()->asFlatVector<StringView>(); + auto doc = arrayParser_.iterate(jsonString); + vector_size_t start = elements->size(); + vector_size_t count = 0; + for (auto arrayElement : doc.get_array()) { + auto s_res = simdjson::to_json_string(arrayElement); + if (s_res.error()) { + continue; + } + const std::string& s = s_res.value(); + elements->resize(start + count + 1); + elements->set(start + count, StringView{s}); + ++count; + } + arrayVector->setOffsetAndSize(vectorIndex, start, count); + arrayVector->setNull(vectorIndex, false); break; }
28-36: Use vector_size_t for indices and avoid map copyPrevent narrowing and reduce work by referencing the per-row map.
- for (int vectorIndex : rows) { + for (vector_size_t vectorIndex : rows) { vector->setNull(vectorIndex, true); if (!isResolved_) { continue; } auto& logEvent = filteredLogEvents_->at(vectorIndex); // TODO: also need to support auto-generated keys - auto userGenNodeIdValueMap = logEvent->get_user_gen_node_id_value_pairs(); - auto const value_it{userGenNodeIdValueMap.find(nodeId_)}; - if (userGenNodeIdValueMap.end() == value_it || + const auto& userGenNodeIdValueMap = + logEvent->get_user_gen_node_id_value_pairs(); + const auto value_it = userGenNodeIdValueMap.find(nodeId_); + if (userGenNodeIdValueMap.end() == value_it || false == value_it->second.has_value()) { continue; }velox/connectors/clp/search_lib/ir/ClpIrCursor.h (3)
17-22: Include the required standard headers; don’t rely on transitive includes.Missing headers for std::function, std::memory, strings, containers, etc., can break downstream builds.
Apply:
#pragma once +#include <functional> +#include <memory> +#include <string> +#include <string_view> +#include <unordered_map> +#include <utility> +#include <vector> + #include "ffi/ir_stream/Deserializer.hpp" #include "velox/connectors/clp/search_lib/BaseClpCursor.h" #include "velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h"
23-23: Forward-declare clp::ReaderInterface to satisfy pointer usage.Avoid a hard dependency and fix compilation.
-namespace facebook::velox::connector::clp::search_lib { +namespace clp { +class ReaderInterface; +} // namespace clp + +namespace facebook::velox::connector::clp::search_lib {
27-32: Invalid ‘this’ capture in NSDMI; move lambda initialisation into the ctor.C++ forbids referencing/capturing ‘this’ in non-static data member initialisers. Also safer to store the callback without NSDMI.
explicit ClpIrCursor( clp_s::InputSource inputSource, std::string_view splitPath, bool ignoreCase) - : BaseClpCursor(inputSource, splitPath), ignoreCase_(ignoreCase) {} + : BaseClpCursor(inputSource, splitPath), ignoreCase_(ignoreCase) { + projectionResolutionCallback_ = + [this]( + [[maybe_unused]] bool /*isAutoGenerated*/, + [[maybe_unused]] ::clp::ffi::SchemaTree::Node::id_t nodeId, + [[maybe_unused]] std::pair<std::string_view, size_t> projected_key_and_index) + -> ystdlib::error_handling::Result<void> { + projectedColumnIdNodeIdMap_.insert( + {projected_key_and_index.second, nodeId}); + return ystdlib::error_handling::success(); + }; + } @@ std::function<ystdlib::error_handling::Result<void>( bool, ::clp::ffi::SchemaTree::Node::id_t, std::pair<std::string_view, size_t>)> - projectionResolutionCallback_ = - [this]( - [[maybe_unused]] bool isAutoGenerated, - [[maybe_unused]] ::clp::ffi::SchemaTree::Node::id_t nodeId, - [[maybe_unused]] std::pair<std::string_view, size_t> - projected_key_and_index) - -> ystdlib::error_handling::Result<void> { - projectedColumnIdNodeIdMap_.insert( - {projected_key_and_index.second, nodeId}); - return ystdlib::error_handling::success(); - }; + projectionResolutionCallback_;Also applies to: 51-65
velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp (8)
24-25: Remove global using-directive; qualify clp_s symbols.Prevents namespace pollution and accidental symbol capture.
-using namespace clp_s; +// Avoid global using-directives. @@ - auto networkAuthOption = inputSource_ == InputSource::Filesystem - ? NetworkAuthOption{.method = AuthMethod::None} - : NetworkAuthOption{.method = AuthMethod::S3PresignedUrlV4}; + auto networkAuthOption = inputSource_ == clp_s::InputSource::Filesystem + ? clp_s::NetworkAuthOption{.method = clp_s::AuthMethod::None} + : clp_s::NetworkAuthOption{.method = clp_s::AuthMethod::S3PresignedUrlV4}; @@ - auto irPath = Path{.source = inputSource_, .path = splitPath_}; - irReader_ = try_create_reader(irPath, networkAuthOption); + auto irPath = clp_s::Path{.source = inputSource_, .path = splitPath_}; + irReader_ = clp_s::try_create_reader(irPath, networkAuthOption);Also applies to: 71-74, 89-91
41-51: Don’t abort on deserialize error; set errorCode_ and return 0.Aligns with archive path; lets pipeline handle failures gracefully.
auto deserializeResult = deserialize(numRows); if (deserializeResult.has_error()) { auto error = deserializeResult.error(); - VELOX_FAIL( - "IR file {} might be broken, failed to deserialize. {}: {}", - this->splitPath_, - error.category().name(), - error.message()); + VLOG(2) << "IR deserialize failed for \"" << splitPath_ << "\": " + << error.category().name() << ": " << error.message(); + errorCode_ = ErrorCode::InternalError; + return 0; } - return irDeserializer_->get_num_log_events_deserialized(); + return getNumFilteredRows(); // Return rows scanned if/when available.
70-76: Reset per-split state in loadSplit().Clear the projection map and reset reader index to avoid stale state across splits.
ErrorCode ClpIrCursor::loadSplit() { + projectedColumnIdNodeIdMap_.clear(); + readerIndex_ = 0;
111-146: Unify qualifier for literal_type_bitmask_t and fix Timestamp mapping.
- Use fully-qualified clp_s::… type to match the header.
- Timestamp isn’t populated yet (per PR); don’t claim EpochDateT.
-std::vector<std::pair<std::string, search::ast::literal_type_bitmask_t>> +std::vector<std::pair<std::string, clp_s::search::ast::literal_type_bitmask_t>> ClpIrCursor::splitFieldsToNamesAndTypes() const { - auto result = std::vector< - std::pair<std::string, search::ast::literal_type_bitmask_t>>{}; + auto result = std::vector< + std::pair<std::string, clp_s::search::ast::literal_type_bitmask_t>>{}; @@ - search::ast::literal_type_bitmask_t literalType; + clp_s::search::ast::literal_type_bitmask_t literalType; switch (column.type) { case ColumnType::Array: - literalType = search::ast::LiteralType::ArrayT; + literalType = clp_s::search::ast::LiteralType::ArrayT; break; case ColumnType::Boolean: - literalType = search::ast::LiteralType::BooleanT; + literalType = clp_s::search::ast::LiteralType::BooleanT; break; case ColumnType::Float: - literalType = search::ast::LiteralType::FloatT; + literalType = clp_s::search::ast::LiteralType::FloatT; break; case ColumnType::Integer: - literalType = search::ast::LiteralType::IntegerT; + literalType = clp_s::search::ast::LiteralType::IntegerT; break; case ColumnType::String: - literalType = search::ast::LiteralType::VarStringT | - search::ast::LiteralType::ClpStringT; + literalType = clp_s::search::ast::LiteralType::VarStringT | + clp_s::search::ast::LiteralType::ClpStringT; break; case ColumnType::Timestamp: - // TODO: IR timestamp support pending; constrain to Unknown to avoid - // mismatched projections. - literalType = search::ast::LiteralType::EpochDateT; + // TODO: IR timestamp support pending; keep Unknown to avoid mismatched projections. + literalType = clp_s::search::ast::LiteralType::UnknownT; break; default: - literalType = search::ast::LiteralType::UnknownT; + literalType = clp_s::search::ast::LiteralType::UnknownT; break; }
17-23: Compare std::error_code properly; include <system_error>.Use make_error_code to match the category and add the header.
+#include <system_error> #include "clp_s/ColumnReader.hpp" #include "clp_s/InputConfig.hpp" @@ - if (std::errc::result_out_of_range == error || + if (error == std::make_error_code(std::errc::result_out_of_range) || irDeserializer_->is_stream_completed()) {Also applies to: 158-160
173-179: Reserve children capacity for ROW vectors.Avoids reallocations.
if (vectorType->kind() == TypeKind::ROW) { - std::vector<VectorPtr> children; + std::vector<VectorPtr> children; + children.reserve(vectorType->as<TypeKind::ROW>().size()); auto& rowType = vectorType->as<TypeKind::ROW>();
191-194: Initialise projectedColumnNodeId for the unresolved path.Prevents passing an indeterminate value to the loader.
- ::clp::ffi::SchemaTree::Node::id_t projectedColumnNodeId; + ::clp::ffi::SchemaTree::Node::id_t projectedColumnNodeId{}; if (isResolved) { projectedColumnNodeId = it->second; }
61-67: Allow unresolved projections; don’t require a 1:1 mapping.Equality assert crashes when some projected columns don’t resolve; we should marshal nulls instead.
- VELOX_CHECK_EQ( - projectedColumnIdNodeIdMap_.size(), - outputColumns_.size(), - "Projected columns size {} does not match fields size {}", - projectedColumnIdNodeIdMap_.size(), - outputColumns_.size()); + VELOX_CHECK_LE( + projectedColumnIdNodeIdMap_.size(), + outputColumns_.size(), + "More projections resolved ({}) than requested fields ({}).", + projectedColumnIdNodeIdMap_.size(), + outputColumns_.size());
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
📥 Commits
Reviewing files that changed from the base of the PR and between ef62d9be4891ef5b64afd4d02bc86861593d78dd and 3962325c2f6780d1e6964a3608c147bf027a4d1f.
📒 Files selected for processing (7)
velox/connectors/clp/search_lib/ir/CMakeLists.txt(1 hunks)velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp(1 hunks)velox/connectors/clp/search_lib/ir/ClpIrCursor.h(1 hunks)velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h(1 hunks)velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cpp(1 hunks)velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h(1 hunks)velox/connectors/clp/tests/ClpConnectorTest.cpp(4 hunks)
🧰 Additional context used
🧠 Learnings (4)
📓 Common learnings
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h:36-39
Timestamp: 2025-08-20T18:00:51.241Z
Learning: In ClpArchiveCursor.h, the clp_s::InputSource type is available through transitive inclusion via BaseClpCursor.h, which includes "clp_s/InputConfig.hpp" where InputSource is defined. The constructor parameter type is properly accessible without needing direct inclusion.
📚 Learning: 2025-08-20T18:13:32.442Z
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/ClpVectorLoader.cpp:24-24
Timestamp: 2025-08-20T18:13:32.442Z
Learning: In ClpVectorLoader.cpp, the include of BaseClpCursor.h is necessary because the implementation uses ColumnType enum values (like ColumnType::Integer, ColumnType::Float) in a switch statement. While ClpVectorLoader.h has a forward declaration of ColumnType, the full definition from BaseClpCursor.h is required in the implementation file to access the enum values.
Applied to files:
velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.hvelox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cppvelox/connectors/clp/search_lib/ir/ClpIrCursor.cppvelox/connectors/clp/search_lib/ir/ClpIrCursor.h
📚 Learning: 2025-08-20T18:00:51.241Z
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h:36-39
Timestamp: 2025-08-20T18:00:51.241Z
Learning: In ClpArchiveCursor.h, the clp_s::InputSource type is available through transitive inclusion via BaseClpCursor.h, which includes "clp_s/InputConfig.hpp" where InputSource is defined. The constructor parameter type is properly accessible without needing direct inclusion.
Applied to files:
velox/connectors/clp/search_lib/ir/ClpIrCursor.h
📚 Learning: 2025-08-20T16:58:56.256Z
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp:32-35
Timestamp: 2025-08-20T16:58:56.256Z
Learning: In the BaseClpCursor class, the constructor takes std::string_view splitPath but stores it as an owning std::string member (splitPath_) using explicit conversion: splitPath_(std::string(splitPath)). This pattern safely avoids dangling reference issues when accepting string_view parameters.
Applied to files:
velox/connectors/clp/search_lib/ir/ClpIrCursor.h
🧬 Code graph analysis (4)
velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h (3)
velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h (1)
velox/vector/LazyVector.h (1)
VectorLoader(177-210)velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cpp (2)
loadInternal(22-133)loadInternal(22-26)
velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp (2)
velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h (1)
readerIndex_(64-64)velox/connectors/clp/search_lib/BaseClpCursor.h (1)
currentSplitLoaded_(122-135)
velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h (2)
velox/connectors/clp/search_lib/ir/ClpIrCursor.h (1)
velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h (1)
velox/connectors/clp/search_lib/ir/ClpIrCursor.h (4)
velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h (2)
ClpIrUnitHandler(29-47)velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h (1)
velox/connectors/clp/search_lib/BaseClpCursor.h (1)
BaseClpCursor(68-137)velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp (14)
fetchNext(28-51)fetchNext(28-28)getNumFilteredRows(53-55)getNumFilteredRows(53-53)createVector(57-68)createVector(57-60)loadSplit(70-109)loadSplit(70-70)splitFieldsToNamesAndTypes(111-147)splitFieldsToNamesAndTypes(112-112)deserialize(149-167)deserialize(149-150)createVectorHelper(169-206)createVectorHelper(169-172)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: Build with GCC / Ubuntu release with resolve_dependency
There was a problem hiding this comment.
🛠️ Refactor suggestion
Fix self-include and complete standard includes
This header includes itself, causing recursive inclusion; also missing and has an unused Exceptions include.
-#include <memory>
-#include <vector>
+#include <cstddef>
+#include <memory>
+#include <vector>
-#include "clp_s/SchemaTree.hpp"
-#include "ffi/ir_stream/Deserializer.hpp"
-#include "velox/common/base/Exceptions.h"
-#include "velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h"
+#include "clp_s/SchemaTree.hpp"
+#include "ffi/ir_stream/Deserializer.hpp"📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| #include <memory> | |
| #include <vector> | |
| #include "clp_s/SchemaTree.hpp" | |
| #include "ffi/ir_stream/Deserializer.hpp" | |
| #include "velox/common/base/Exceptions.h" | |
| #include "velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h" | |
| #include <cstddef> | |
| #include <memory> | |
| #include <vector> | |
| #include "clp_s/SchemaTree.hpp" | |
| #include "ffi/ir_stream/Deserializer.hpp" |
🤖 Prompt for AI Agents
In velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h around lines 19 to 26,
the header mistakenly includes itself and omits the <cstddef> header while
containing an unused Exceptions include; remove the self-include line
("velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h") and the unused
"velox/common/base/Exceptions.h" include, add #include <cstddef> alongside the
existing <memory> and <vector> includes, and ensure only the necessary headers
("clp_s/SchemaTree.hpp" and "ffi/ir_stream/Deserializer.hpp") remain.
3962325 to
36c02c7
Compare
gibber9809
left a comment
gibber9809
left a comment
There was a problem hiding this comment.
For code in the IR subdirectory: LGTM!
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (17)
velox/connectors/clp/search_lib/ir/CMakeLists.txt (1)
23-33: Link against the IR stream library (not archive reader).This target implements IR; linking PUBLIC to clp_s::archive_reader is wrong and risks unresolved symbols at link time. Switch to the IR stream target exported by clp_s (name may vary).
velox_link_libraries( clp-s-ir-search - PUBLIC clp_s::archive_reader + PUBLIC clp_s::ir_stream # Adjust if your clp_s export name differs (e.g., clp_s::ir or clp_s::ir_reader) PRIVATE # Once the IR-stream-related targets are exported, this should be updated to # use more fine-grained dependencies. clp_s::clp_dependencies clp_s::io clp_s::search clp_s::search::kql velox_vector)velox/connectors/clp/tests/ClpConnectorTest.cpp (1)
75-81: Stabilize order-dependent assertions: force single-driver execution.Without this, multi-driver runs can reorder rows and cause flaky tests.
RowVectorPtr getResults( const core::PlanNodePtr& planNode, std::vector<exec::Split>&& splits) { return exec::test::AssertQueryBuilder(planNode) + .maxDrivers(1) // Ensure deterministic ordering for order-sensitive assertions. .splits(std::move(splits)) .copyResults(pool()); }velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cpp (4)
22-27: Honour the ValueHook contract (or fail fast).
loadInternalignoreshook; this violates theVectorLoadercontract and can defeat pushdown.void ClpIrVectorLoader::loadInternal( RowSet rows, ValueHook* hook, vector_size_t resultSize, VectorPtr* result) { + if (hook != nullptr) { + VELOX_UNSUPPORTED("ClpIrVectorLoader does not support ValueHook-based loading yet."); + } + (void)resultSize; auto vector = *result;
28-36: Use correct index type and avoid copying the map.
- Iterate with
vector_size_t.- Take a const reference to avoid copying the per-row map.
- for (int vectorIndex : rows) { + for (vector_size_t vectorIndex : rows) { vector->setNull(vectorIndex, true); if (!isResolved_) { continue; } auto& logEvent = filteredLogEvents_->at(vectorIndex); // TODO: also need to support auto-generated keys - auto userGenNodeIdValueMap = logEvent->get_user_gen_node_id_value_pairs(); + const auto& userGenNodeIdValueMap = + logEvent->get_user_gen_node_id_value_pairs();
92-94: Prefer Velox RTTI helper over shared_ptr RTTI.- case ColumnType::Array: { - auto arrayVector = std::dynamic_pointer_cast<ArrayVector>(vector); + case ColumnType::Array: { + auto* arrayVector = vector->as<ArrayVector>();
113-126: Fix ArrayVector loading: wrong offsets and dangling views.
- Offsets are hard-coded to 0, corrupting data across rows.
- Storing
std::string_viewto temporaries fromto_json_string(...).value()creates dangling views.- size_t numElements{0ULL}; - auto elements = arrayVector->elements()->asFlatVector<StringView>(); - auto obj = arrayParser_.iterate(jsonString); - std::vector<std::string_view> rawElements; - for (auto arrayElement : obj.get_array()) { - auto raw_element = simdjson::to_json_string(arrayElement).value(); - rawElements.emplace_back(raw_element); - } - elements->resize(rawElements.size()); - for (auto& raw_element : rawElements) { - elements->set(numElements++, StringView(raw_element)); - } - arrayVector->setOffsetAndSize(vectorIndex, 0ULL, numElements); + auto* elements = arrayVector->elements()->asFlatVector<StringView>(); + auto doc = arrayParser_.iterate(jsonString); + const vector_size_t start = elements->size(); + vector_size_t count{0}; + for (auto elem : doc.get_array()) { + auto sres = simdjson::to_json_string(elem); + if (sres.error()) { + continue; // skip malformed element + } + elements->resize(start + count + 1); + elements->set(start + count, StringView(sres.value())); + ++count; + } + arrayVector->setOffsetAndSize(vectorIndex, start, count); arrayVector->setNull(vectorIndex, false);velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h (2)
21-25: Add direct include for VectorLoaderThe class derives from
VectorLoaderbut relies on transitive inclusion throughLazyVector.h. Add a direct include to avoid fragility.#include "connectors/clp/search_lib/BaseClpCursor.h" #include "ffi/ir_stream/Deserializer.hpp" +#include "velox/vector/VectorLoader.h" #include "velox/vector/FlatVector.h" #include "velox/vector/LazyVector.h"
28-40: Document thread-safety guaranteesThe class uses an instance member
arrayParser_which makes it not thread-safe. Document that instances are not thread-safe andVectorLoader::loadshould be called at most once per instance.Add a class-level comment:
/** * ClpIrVectorLoader loads vectors from CLP IR log events. * * Thread-safety: Instances are not thread-safe and must not be used * concurrently. The VectorLoader::load method is intended to be called * at most once per instance due to stateful arrayParser_. */ class ClpIrVectorLoader : public VectorLoader {velox/connectors/clp/search_lib/ir/ClpIrCursor.h (5)
19-22: Add missing standard library headersThe header uses standard library types without including their headers. Add the required includes to avoid build failures when transitive includes change.
#pragma once +#include <functional> +#include <memory> +#include <string> +#include <string_view> +#include <unordered_map> +#include <utility> + #include "ffi/ir_stream/Deserializer.hpp" #include "velox/connectors/clp/search_lib/BaseClpCursor.h" #include "velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h"
19-22: Forward-declare clp::ReaderInterfaceThe class uses
std::shared_ptr<::clp::ReaderInterface>at line 72 without including or forward-declaring the type.#include "velox/connectors/clp/search_lib/BaseClpCursor.h" #include "velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h" +namespace clp { +class ReaderInterface; +} // namespace clp + namespace facebook::velox::connector::clp::search_lib {
51-65: Invalid use of 'this' in non-static data member initializerC++ doesn't allow capturing 'this' in a non-static data member initializer. Move the lambda initialization to the constructor.
explicit ClpIrCursor( clp_s::InputSource inputSource, std::string_view splitPath, bool ignoreCase) - : BaseClpCursor(inputSource, splitPath), ignoreCase_(ignoreCase) {} + : BaseClpCursor(inputSource, splitPath), ignoreCase_(ignoreCase) { + projectionResolutionCallback_ = + [this]( + [[maybe_unused]] bool isAutoGenerated, + [[maybe_unused]] ::clp::ffi::SchemaTree::Node::id_t nodeId, + [[maybe_unused]] std::pair<std::string_view, size_t> + projected_key_and_index) + -> ystdlib::error_handling::Result<void> { + projectedColumnIdxNodeIdMap_.insert( + {projected_key_and_index.second, nodeId}); + return ystdlib::error_handling::success(); + }; + } // ... rest of class ... std::function<ystdlib::error_handling::Result<void>( bool, ::clp::ffi::SchemaTree::Node::id_t, std::pair<std::string_view, size_t>)> - projectionResolutionCallback_ = - [this]( - [[maybe_unused]] bool isAutoGenerated, - [[maybe_unused]] ::clp::ffi::SchemaTree::Node::id_t nodeId, - [[maybe_unused]] std::pair<std::string_view, size_t> - projected_key_and_index) - -> ystdlib::error_handling::Result<void> { - projectedColumnIdxNodeIdMap_.insert( - {projected_key_and_index.second, nodeId}); - return ystdlib::error_handling::success(); - }; + projectionResolutionCallback_;
77-79: Inconsistent type qualifier for literal_type_bitmask_tThe header declares the return type with
clp_s::search::ast::literal_type_bitmask_tbut the implementation file usessearch::ast::literal_type_bitmask_t. Ensure consistency.Verify and update the implementation file to use the fully-qualified type:
std::vector<std::pair<std::string, clp_s::search::ast::literal_type_bitmask_t>> ClpIrCursor::splitFieldsToNamesAndTypes() const {
73-74: Clear projection map when loading a new splitThe
projectedColumnIdxNodeIdMap_should be cleared when loading a new split to prevent stale mappings across splits.In the
loadSplit()implementation, add:ErrorCode ClpIrCursor::loadSplit() { projectedColumnIdxNodeIdMap_.clear(); readerIndex_ = 0; // ... rest of implementationvelox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h (4)
19-26: Fix self-include and add missing standard headerThe header includes itself at line 25, which will cause a recursive include error. Also missing
<cstddef>forsize_t.#include <memory> #include <vector> +#include <cstddef> #include "clp_s/SchemaTree.hpp" #include "ffi/ir_stream/Deserializer.hpp" -#include "velox/common/base/Exceptions.h" -#include "velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h"
31-34: Use member initializer list for cleaner initializationInitialize
filteredLogEvents_directly in the member initializer list rather than in the constructor body.- ClpIrUnitHandler() { - filteredLogEvents_ = std::make_shared< - std::vector<std::unique_ptr<::clp::ffi::KeyValuePairLogEvent>>>(); - } + ClpIrUnitHandler() + : filteredLogEvents_{std::make_shared< + std::vector<std::unique_ptr<::clp::ffi::KeyValuePairLogEvent>>>()} {}
40-42: Mark unused parameter to suppress warningThe
log_event_idxparameter is unused and should be marked as such.[[nodiscard]] auto handle_log_event( ::clp::ffi::KeyValuePairLogEvent log_event, - size_t log_event_idx) -> ::clp::ffi::ir_stream::IRErrorCode { + [[maybe_unused]] size_t log_event_idx) -> ::clp::ffi::ir_stream::IRErrorCode {
49-54: Mark trivial handlers as noexceptThese methods perform trivial operations and don't throw. Mark them
noexceptto tighten the exception contract.[[nodiscard]] auto handle_utc_offset_change( [[maybe_unused]] ::clp::UtcOffset utc_offset_old, [[maybe_unused]] ::clp::UtcOffset utc_offset_new) - -> ::clp::ffi::ir_stream::IRErrorCode { + noexcept -> ::clp::ffi::ir_stream::IRErrorCode { return ::clp::ffi::ir_stream::IRErrorCode::IRErrorCode_Success; } // ... - [[nodiscard]] auto handle_end_of_stream() - -> ::clp::ffi::ir_stream::IRErrorCode { + [[nodiscard]] auto handle_end_of_stream() noexcept + -> ::clp::ffi::ir_stream::IRErrorCode { return ::clp::ffi::ir_stream::IRErrorCode::IRErrorCode_Success; }Also applies to: 65-68
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
📥 Commits
Reviewing files that changed from the base of the PR and between 3962325c2f6780d1e6964a3608c147bf027a4d1f and ee14262.
📒 Files selected for processing (7)
velox/connectors/clp/search_lib/ir/CMakeLists.txt(1 hunks)velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp(1 hunks)velox/connectors/clp/search_lib/ir/ClpIrCursor.h(1 hunks)velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h(1 hunks)velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cpp(1 hunks)velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h(1 hunks)velox/connectors/clp/tests/ClpConnectorTest.cpp(4 hunks)
🧰 Additional context used
🧠 Learnings (4)
📓 Common learnings
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h:36-39
Timestamp: 2025-08-20T18:00:51.241Z
Learning: In ClpArchiveCursor.h, the clp_s::InputSource type is available through transitive inclusion via BaseClpCursor.h, which includes "clp_s/InputConfig.hpp" where InputSource is defined. The constructor parameter type is properly accessible without needing direct inclusion.
📚 Learning: 2025-08-20T18:13:32.442Z
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/ClpVectorLoader.cpp:24-24
Timestamp: 2025-08-20T18:13:32.442Z
Learning: In ClpVectorLoader.cpp, the include of BaseClpCursor.h is necessary because the implementation uses ColumnType enum values (like ColumnType::Integer, ColumnType::Float) in a switch statement. While ClpVectorLoader.h has a forward declaration of ColumnType, the full definition from BaseClpCursor.h is required in the implementation file to access the enum values.
Applied to files:
velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cppvelox/connectors/clp/search_lib/ir/ClpIrCursor.hvelox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h
📚 Learning: 2025-08-20T18:00:51.241Z
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h:36-39
Timestamp: 2025-08-20T18:00:51.241Z
Learning: In ClpArchiveCursor.h, the clp_s::InputSource type is available through transitive inclusion via BaseClpCursor.h, which includes "clp_s/InputConfig.hpp" where InputSource is defined. The constructor parameter type is properly accessible without needing direct inclusion.
Applied to files:
velox/connectors/clp/search_lib/ir/ClpIrCursor.h
📚 Learning: 2025-08-20T16:58:56.256Z
Learnt from: anlowee
PR: y-scope/velox#25
File: velox/connectors/clp/search_lib/archive/ClpArchiveCursor.cpp:32-35
Timestamp: 2025-08-20T16:58:56.256Z
Learning: In the BaseClpCursor class, the constructor takes std::string_view splitPath but stores it as an owning std::string member (splitPath_) using explicit conversion: splitPath_(std::string(splitPath)). This pattern safely avoids dangling reference issues when accepting string_view parameters.
Applied to files:
velox/connectors/clp/search_lib/ir/ClpIrCursor.h
🧬 Code graph analysis (4)
velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h (2)
velox/connectors/clp/search_lib/ir/ClpIrCursor.h (1)
velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h (1)
velox/connectors/clp/search_lib/ir/ClpIrCursor.h (4)
velox/connectors/clp/search_lib/ir/ClpIrUnitHandler.h (2)
ClpIrUnitHandler(29-47)velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h (1)
velox/connectors/clp/search_lib/BaseClpCursor.h (1)
BaseClpCursor(68-137)velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp (14)
fetchNext(28-51)fetchNext(28-28)getNumFilteredRows(53-55)getNumFilteredRows(53-53)createVector(57-68)createVector(57-60)loadSplit(70-109)loadSplit(70-70)splitFieldsToNamesAndTypes(111-147)splitFieldsToNamesAndTypes(112-112)deserialize(149-170)deserialize(149-150)createVectorHelper(172-209)createVectorHelper(172-175)
velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h (2)
velox/vector/LazyVector.h (1)
VectorLoader(177-210)velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.cpp (2)
loadInternal(22-133)loadInternal(22-26)
velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp (2)
velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h (1)
readerIndex_(64-64)velox/connectors/clp/search_lib/BaseClpCursor.h (1)
currentSplitLoaded_(122-135)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: Build with GCC / Ubuntu release with resolve_dependency
🔇 Additional comments (11)
velox/connectors/clp/search_lib/ir/ClpIrCursor.cpp (11)
24-24: Avoid global using-directive; qualify clp_s symbols to prevent namespace bleed.
Remove the file-scope using and fully-qualify affected identifiers.
[ suggest_nitpick ][ duplicate_comment ]-using namespace clp_s; +// Avoid global using-directives; qualify clp_s symbols explicitly.- auto networkAuthOption = inputSource_ == InputSource::Filesystem - ? NetworkAuthOption{.method = AuthMethod::None} - : NetworkAuthOption{.method = AuthMethod::S3PresignedUrlV4}; + auto networkAuthOption = inputSource_ == clp_s::InputSource::Filesystem + ? clp_s::NetworkAuthOption{.method = clp_s::AuthMethod::None} + : clp_s::NetworkAuthOption{.method = clp_s::AuthMethod::S3PresignedUrlV4};- auto irPath = Path{.source = inputSource_, .path = splitPath_}; - irReader_ = try_create_reader(irPath, networkAuthOption); + auto irPath = clp_s::Path{.source = inputSource_, .path = splitPath_}; + irReader_ = clp_s::try_create_reader(irPath, networkAuthOption);Also applies to: 71-74, 89-91
41-49: Don’t hard-crash on IR deserialize error; set errorCode_ and return 0.
Align with archive path; let the pipeline handle the failure.
[ raise_critical_issue ][ suggest_essential_refactor ][ duplicate_comment ]auto deserializeResult = deserialize(numRows); if (deserializeResult.has_error()) { auto error = deserializeResult.error(); - VELOX_FAIL( - "IR file {} might be broken, failed to deserialize. {}: {}", - this->splitPath_, - error.category().name(), - error.message()); + VLOG(2) << "IR deserialize failed for \"" << splitPath_ << "\": " + << error.category().name() << ": " << error.message(); + errorCode_ = ErrorCode::InternalError; + return 0; }
50-51: Confirm fetchNext()’s return semantics (rows scanned per-batch).
If BaseClpCursor expects “rows scanned this batch,” ensure this isn’t cumulative. If cumulative, consider returning the per-batch count (e.g., from deserialize).
[ request_verification ]Run to verify contract and usages:
#!/bin/bash # Inspect BaseClpCursor and Archive cursor behaviour. rg -n -C3 'class\s+BaseClpCursor\b' velox rg -n -C3 'virtual\s+uint64_t\s+fetchNext\s*\(' velox rg -n -C3 'ClpArchiveCursor::fetchNext' velox rg -n -C3 'getNumFilteredRows\s*\(' velox
61-67: Do not assert all projected columns are resolved; unresolved must marshal nulls.
This check can crash before resolution completes.
[ raise_critical_issue ][ duplicate_comment ]- VELOX_CHECK_EQ( - projectedColumnIdxNodeIdMap_.size(), - outputColumns_.size(), - "Projected columns size {} does not match fields size {}", - projectedColumnIdxNodeIdMap_.size(), - outputColumns_.size()); + VELOX_CHECK_LE( + projectedColumnIdxNodeIdMap_.size(), + outputColumns_.size(), + "Projected columns map larger than fields: {} > {}", + projectedColumnIdxNodeIdMap_.size(), + outputColumns_.size());
136-139: Timestamp advertised as unsupported; don’t expose EpochDateT.
Map to UnknownT (or gate by capability) to avoid mismatched projections.
[ suggest_essential_refactor ][ duplicate_comment ]- case ColumnType::Timestamp: - // TODO: IR timestamp support pending; constrain to Unknown to avoid - // mismatched projections. - literalType = search::ast::LiteralType::EpochDateT; + case ColumnType::Timestamp: + // IR timestamp support pending; avoid projecting as EpochDateT. + literalType = search::ast::LiteralType::UnknownT; break;
115-117: Avoid copy of Field; use const reference.
[ suggest_nitpick ]- auto column = outputColumns_[i]; + const auto& column = outputColumns_[i];
17-23: Include <system_error> for std::make_error_code.
[ suggest_nitpick ][ duplicate_comment ]#include "clp_s/ColumnReader.hpp" #include "clp_s/InputConfig.hpp" #include "ffi/ir_stream/search/QueryHandler.hpp" +#include <system_error>
158-159: Compare std::error_code to std::errc via make_error_code.
[ suggest_essential_refactor ][ duplicate_comment ]- if (std::errc::result_out_of_range == error || + if (error == std::make_error_code(std::errc::result_out_of_range) || irDeserializer_->is_stream_completed()) {
176-182: Pre-reserve children capacity for ROW to avoid reallocations.
[ suggest_nitpick ][ duplicate_comment ]if (vectorType->kind() == TypeKind::ROW) { std::vector<VectorPtr> children; auto& rowType = vectorType->as<TypeKind::ROW>(); + children.reserve(rowType.size()); for (uint32_t i = 0; i < rowType.size(); ++i) { children.push_back( createVectorHelper(pool, rowType.childAt(i), vectorSize)); }
194-197: Initialise projectedColumnNodeId when unresolved to avoid UB.
[ raise_critical_issue ][ duplicate_comment ]- ::clp::ffi::SchemaTree::Node::id_t projectedColumnNodeId; + ::clp::ffi::SchemaTree::Node::id_t projectedColumnNodeId{}; if (isResolved) { projectedColumnNodeId = it->second; }
186-209: Verify LazyVector loader lifetime vs. filtered-log-events buffer reset.
Filtered events are cleared at the start of deserialize(); ensure all LazyVectors created for a batch are loaded before the next fetch to prevent use-after-clear.
[ request_verification ]If needed, add a comment/guard to guarantee load-before-next-fetch or pass an owning snapshot to ClpIrVectorLoader.
| auto stringVector = vector->asFlatVector<StringView>(); | ||
| if (value->is<std::string>()) { | ||
| auto stringValue = value->get_immutable_view<std::string>(); | ||
| stringVector->set(vectorIndex, StringView(stringValue)); | ||
| } else if (value->is<::clp::ir::EightByteEncodedTextAst>()) { | ||
| auto decodeResult = | ||
| value->get_immutable_view<::clp::ir::EightByteEncodedTextAst>() | ||
| .decode_and_unparse(); | ||
| if (!decodeResult.has_value()) { | ||
| continue; | ||
| } | ||
| stringVector->set(vectorIndex, StringView(decodeResult.value())); | ||
| } else if (value->is<::clp::ir::FourByteEncodedTextAst>()) { | ||
| auto decodeResult = | ||
| value->get_immutable_view<::clp::ir::FourByteEncodedTextAst>() | ||
| .decode_and_unparse(); | ||
| if (!decodeResult.has_value()) { | ||
| continue; | ||
| } | ||
| stringVector->set(vectorIndex, StringView(decodeResult.value())); | ||
| } else { | ||
| continue; | ||
| } | ||
| vector->setNull(vectorIndex, false); |
There was a problem hiding this comment.
🧹 Nitpick (assertive)
Deduplicate AST decode paths with a small helper.
Both String and Array cases repeat decode_and_unparse() for 4/8-byte ASTs. Factor into a local lambda to reduce branching and bugs.
Example sketch (apply near the top of the function scope):
auto decodeAstToString = [&](const auto& v) -> std::optional<std::string> {
if (v->is<::clp::ir::EightByteEncodedTextAst>()) {
return v->get_immutable_view<::clp::ir::EightByteEncodedTextAst>().decode_and_unparse();
}
if (v->is<::clp::ir::FourByteEncodedTextAst>()) {
return v->get_immutable_view<::clp::ir::FourByteEncodedTextAst>().decode_and_unparse();
}
return std::nullopt;
};Then reuse in both switch arms.
Also applies to: 95-111
|
|
||
| #pragma once | ||
|
|
||
| #include <simdjson.h> |
There was a problem hiding this comment.
Include missing standard headers
This header uses std::shared_ptr but doesn't include <memory>. Add the required standard library headers to avoid relying on transitive includes.
#include <simdjson.h>
+#include <memory>
+#include <vector>
#include "connectors/clp/search_lib/BaseClpCursor.h"📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| #include <simdjson.h> | |
| #include <simdjson.h> | |
| #include <memory> | |
| #include <vector> | |
| #include "connectors/clp/search_lib/BaseClpCursor.h" | |
| #include "ffi/ir_stream/Deserializer.hpp" | |
| #include "velox/vector/LazyVector.h" | |
| #include "velox/vector/FlatVector.h" | |
| #include "velox/vector/VectorLoader.h" | |
| #include "velox/type/Timestamp.h" | |
| // …rest of file… |
🤖 Prompt for AI Agents
In velox/connectors/clp/search_lib/ir/ClpIrVectorLoader.h around line 19, the
header uses std::shared_ptr but doesn't include the required <memory> header;
add #include <memory> (and any other missing standard headers it depends on)
near the top of the file so the header doesn't rely on transitive includes.
7ae5119
into
y-scope:presto-0.293-clp-connector
3 participants
Description
This PR is the successor of #25 , and must be merged after #25 is merged.(aleady merged)This PR added the implementation of the
BaseClpCursorfor IR to support search on IR stream. Currently since the auto-generated keys are queried by a leading@in the field name, the Velox doesn't support this special char now. We will figure out how to support query on auto-generated keys in the future. In this PR, we only consider the user-generated keys.This PR doesn't populate the timestamp type yet, will support in the next PR.
This PR doesn't modify the docs, will add in the next PR.
Checklist
breaking change.
Validation performed
Passed the CI.
End-to-end test.
Summary by CodeRabbit
New Features
Refactor
Tests
Chores