feat(clp-s): Support per-archive min and max aggregation output to stdout and the results cache.#2345
Open
davemarco wants to merge 57 commits into
Open
feat(clp-s): Support per-archive min and max aggregation output to stdout and the results cache.#2345davemarco wants to merge 57 commits into
davemarco wants to merge 57 commits into
Conversation
…e results cache. Allows --count and --count-by-time to be used with the results-cache output handler, writing aggregation results directly to MongoDB via _id-keyed atomic $inc upserts so that partial results from multiple search processes merge correctly without a reducer.
…he search usage text outputs to the reducer.
…nation and simplify comments.
…dout. Adds `--count` and `--count-by-time` to the stdout output handler, emitting per-archive results as newline-delimited JSON. The options may be given without naming the `stdout` handler (e.g. `clp-s s <archives> <query> --count`), and the kv-ir search path now reports these aggregations as unsupported, consistent with the reducer and results-cache handlers.
… results cache. Adds --min/--max <field> aggregations, extracting the field's numeric value from each matched record's marshalled JSON. Unifies the results-cache count handlers into a single AggregationToResultsCacheOutputHandler handling count, count-by-time, min, and max.
Contributor
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughAdds variant-based aggregation types, sinks, and CLI parsing for search. Replaces count-specific results-cache handlers with a single aggregation-aware output handler, updates runtime handler selection, and adds exact int/double comparison helpers plus shared MongoDB connection logic. ChangesAggregation Framework
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested reviewers
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
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 |
davidlion
requested changes
Jun 24, 2026
davidlion
left a comment
davidlion
left a comment
Member
There was a problem hiding this comment.
Quick pass to try and speed things up.
| #include "../clp/type_utils.hpp" | ||
| #include "../reducer/types.hpp" | ||
| #include "FileReader.hpp" | ||
| #include "search/ast/SearchUtils.hpp" |
Member
There was a problem hiding this comment.
Suggested change
| #include "search/ast/SearchUtils.hpp" | |
| #include <clp_s/search/ast/SearchUtils.hpp> |
For new include statements let's follow the new guidelines to always use <>.
Member
There was a problem hiding this comment.
This was marked resolved, but the change was never applied...
| std::map<std::string, po::options_description const*> const& subcommands, | ||
| std::vector<std::string> const& options | ||
| std::vector<std::string> const& options, | ||
| std::optional<std::string> const& default_subcommand = std::nullopt |
Member
There was a problem hiding this comment.
Update doc string with this param + update description.
|
|
||
| StdoutOutputHandlerOptions stdout_options{}; | ||
| po::options_description stdout_output_handler_options("Stdout Output Handler Options"); | ||
| // clang-format off |
Member
There was a problem hiding this comment.
How ugly is it if we just let clang-format do it now? I feel like this might be unnecessary since it is likely a from very early on.
…ers. - Take string_view for the results-cache handler uri/collection/dataset/archive_id params and materialize std::string only at the mongocxx call sites. - Drop the redundant ::clp_s:: qualifier on the count handlers' base class. - Rename the count-by-time write timestamp param to timestamp_ms to match the bucket size unit.
…n handler. Merges the results-cache count work and applies the same review fixes to AggregationToStdoutOutputHandler: take string_view for archive_id, drop the redundant ::clp_s:: qualifier on the base class, rename the count-by-time bucket size to count_by_time_bucket_size_ms, and rename the count-by-time write timestamp param to timestamp_ms.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
♻️ Duplicate comments (1)
components/core/src/clp_s/Aggregation.cpp (1)
95-100: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winHandle
number_unsignedbefore casting toint64_t.
is_number_integer()also matches unsigned JSON values here, sonode->get<int64_t>()can turn a value aboveINT64_MAXinto the wrong signed extreme and poison both--minand--maxresults. Please branch onis_number_unsigned(), range-check it, and skip or reject out-of-range values instead of narrowing them throughint64_t.Proposed fix
- Extreme const candidate{ - node->is_number_integer() ? Extreme{node->get<int64_t>()} : Extreme{node->get<double>()} - }; + Extreme candidate; + if (node->is_number_unsigned()) { + auto const value{node->get<uint64_t>()}; + if (value > static_cast<uint64_t>(std::numeric_limits<int64_t>::max())) { + return; + } + candidate = static_cast<int64_t>(value); + } else if (node->is_number_integer()) { + candidate = node->get<int64_t>(); + } else { + candidate = node->get<double>(); + }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@components/core/src/clp_s/Aggregation.cpp` around lines 95 - 100, The numeric handling in `Aggregation::` currently treats unsigned JSON numbers as integers via `is_number_integer()`, which can narrow values above `INT64_MAX` into the wrong `Extreme` and corrupt min/max aggregation. Update the branch that builds `candidate` to handle `is_number_unsigned()` explicitly before `node->get<int64_t>()`, add a range check for values that cannot fit in `int64_t`, and skip or reject those out-of-range values instead of casting them through `int64_t`.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Duplicate comments:
In `@components/core/src/clp_s/Aggregation.cpp`:
- Around line 95-100: The numeric handling in `Aggregation::` currently treats
unsigned JSON numbers as integers via `is_number_integer()`, which can narrow
values above `INT64_MAX` into the wrong `Extreme` and corrupt min/max
aggregation. Update the branch that builds `candidate` to handle
`is_number_unsigned()` explicitly before `node->get<int64_t>()`, add a range
check for values that cannot fit in `int64_t`, and skip or reject those
out-of-range values instead of casting them through `int64_t`.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: 652926cd-0e0c-4348-a1a8-e1596dab4095
📒 Files selected for processing (7)
components/core/src/clp_s/Aggregation.cppcomponents/core/src/clp_s/Aggregation.hppcomponents/core/src/clp_s/AggregationSink.cppcomponents/core/src/clp_s/AggregationSink.hppcomponents/core/src/clp_s/IntFloatCompare.hppcomponents/core/src/clp_s/OutputHandlerImpl.cppcomponents/core/src/clp_s/ResultsCacheUtils.hpp
💤 Files with no reviewable changes (1)
- components/core/src/clp_s/OutputHandlerImpl.cpp
…reducer dispatch, rename aggregation error. - CountByTimeAggregation rejects non-positive bucket sizes. - ResultsCacheSink clears its buffer after a successful flush to avoid duplicate inserts. - Reducer output dispatch handles count-by-time explicitly and throws on unsupported aggregations. - Rename KvIrSearchErrorEnum::CountSupportNotImplemented to AggregationSupportNotImplemented.
…er clang-tidy. Make connect_to_results_cache a non-template function defined in ResultsCacheUtils.cpp instead of a header-only template. The template existed only to throw each caller's nested OperationFailed, but the call sites are caught as std::exception and TraceableException::what() ignores the exception type, so collapsing them into a single ResultsCacheConnectionError is behavior-preserving. This also gives the header a sibling .cpp, so clang-tidy resolves the mongocxx include path (header-only templates fall back to a non-mongo compile command and fail with "mongocxx/client.hpp file not found"). Removes the now-dead nested OperationFailed classes and the reducer parse-time comment.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@components/core/src/clp_s/ResultsCacheUtils.cpp`:
- Around line 20-22: The URI handling in the cache setup currently accepts a
missing database name and still returns a collection handle from
client[mongo_uri.database()], which can create an unusable empty database
reference. Update the helper that builds the mongocxx::uri and client to
validate mongo_uri.database() before accessing the collection, and fail fast
with ErrorCodeBadParamDbUri when the database component is empty. Keep the check
close to the return path so the function only returns a valid
client[database][collection] handle when the URI includes a database name.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: 359ca3f7-68ba-48b4-abd8-9345fb9add80
📒 Files selected for processing (9)
components/core/src/clp_s/AggregationSink.cppcomponents/core/src/clp_s/AggregationSink.hppcomponents/core/src/clp_s/CMakeLists.txtcomponents/core/src/clp_s/IntFloatCompare.hppcomponents/core/src/clp_s/OutputHandlerImpl.cppcomponents/core/src/clp_s/OutputHandlerImpl.hppcomponents/core/src/clp_s/ResultsCacheUtils.cppcomponents/core/src/clp_s/ResultsCacheUtils.hppcomponents/core/src/clp_s/clp-s.cpp
💤 Files with no reviewable changes (3)
- components/core/src/clp_s/clp-s.cpp
- components/core/src/clp_s/AggregationSink.hpp
- components/core/src/clp_s/OutputHandlerImpl.hpp
davidlion
requested changes
Jul 2, 2026
Comment on lines
+22
to
+24
| #include "Aggregation.hpp" | ||
| #include "AggregationSink.hpp" | ||
| #include "CommandLineArguments.hpp" |
Contributor
Author
There was a problem hiding this comment.
switched the new project headers to <clp_s/…>.
| #include "../reducer/network_utils.hpp" | ||
| #include "../reducer/Record.hpp" | ||
| #include "archive_constants.hpp" | ||
| #include "ResultsCacheUtils.hpp" |
Contributor
Author
There was a problem hiding this comment.
switched the new project headers to <clp_s/…>.
Comment on lines
+24
to
+31
| // Destructor | ||
| virtual ~AggregationSink() = default; | ||
|
|
||
| // Explicitly disable copy and move constructor/assignment | ||
| AggregationSink(AggregationSink const&) = delete; | ||
| auto operator=(AggregationSink const&) -> AggregationSink& = delete; | ||
| AggregationSink(AggregationSink&&) = delete; | ||
| auto operator=(AggregationSink&&) -> AggregationSink& = delete; |
Member
There was a problem hiding this comment.
Sorry, this particular change/rule is new and my fault for not getting around to updating the guidelines yet.
Suggested change
| // Destructor | |
| virtual ~AggregationSink() = default; | |
| // Explicitly disable copy and move constructor/assignment | |
| AggregationSink(AggregationSink const&) = delete; | |
| auto operator=(AggregationSink const&) -> AggregationSink& = delete; | |
| AggregationSink(AggregationSink&&) = delete; | |
| auto operator=(AggregationSink&&) -> AggregationSink& = delete; | |
| // Delete copy constructor and assignment operator | |
| AggregationSink(AggregationSink const&) = delete; | |
| auto operator=(AggregationSink const&) -> AggregationSink& = delete; | |
| // Delete move constructor and assignment operator | |
| AggregationSink(AggregationSink&&) = delete; | |
| auto operator=(AggregationSink&&) -> AggregationSink& = delete; | |
| // Destructor | |
| virtual ~AggregationSink() = default; |
Contributor
Author
There was a problem hiding this comment.
Applied the suggested ordering.
davidlion
requested changes
Jul 6, 2026
davidlion
left a comment
davidlion
left a comment
Member
There was a problem hiding this comment.
all minor naming and style changes
| #include <boost/program_options/variables_map.hpp> | ||
|
|
||
| #include "../reducer/types.hpp" | ||
| #include "Aggregation.hpp" |
Comment on lines
+64
to
+65
| static constexpr bool cNeedsMetadata = false; | ||
| static constexpr bool cNeedsMarshalledRecord = false; |
Member
There was a problem hiding this comment.
style nit (shows up in other places too)
Suggested change
| static constexpr bool cNeedsMetadata = false; | |
| static constexpr bool cNeedsMarshalledRecord = false; | |
| static constexpr bool cNeedsMetadata{false}; | |
| static constexpr bool cNeedsMarshalledRecord{false}; |
Comment on lines
+1218
to
+1220
| if (false == m_aggregation.has_value() | ||
| || std::holds_alternative<MinMaxAggregation>(m_aggregation.value())) | ||
| { |
Member
There was a problem hiding this comment.
@coderabbitai Please make a issue to remove the old reducer code that shouldn't be used anymore.
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
gh repo view y-scope/clp --json nameWithOwner -q .nameWithOwner
fd . components/core/src/reducer -t f | head -50
rg -n "Count" components/core/src/reducer -l | head -30Length of output: 2961
🏁 Script executed:
#!/bin/bash
set -euo pipefail
repo="y-scope/clp"
body=$(cat <<'EOF'
## Summary
Follow-up from PR `#2345` (aggregation pipeline refactor for `clp-s`), as requested by `@davidlion` in this comment thread: https://github.com/y-scope/clp/pull/2345#discussion_r3526356325
PR `#2345` introduces a new variant-based `Aggregation`/`AggregationSink` design in `clp_s` (see `components/core/src/clp_s/Aggregation.hpp`, `Aggregation.cpp`, `AggregationSink.hpp`, `AggregationSink.cpp`) and reworks how aggregation results are produced and dispatched to output handlers (`components/core/src/clp_s/OutputHandlerImpl.hpp/.cpp`, `CommandLineArguments.cpp`, `clp-s.cpp`). The reducer output handler path currently still only supports `count` and `count-by-time` aggregations (see the validation added in `CommandLineArguments::parse_reducer_output_handler_options` in `components/core/src/clp_s/CommandLineArguments.cpp`), leaving older reducer-related code that predates this refactor and may no longer be needed now that `clp_s` computes aggregations internally.
## Required changes
- Audit the reducer output handler code path in `clp_s` (`components/core/src/clp_s/CommandLineArguments.cpp`, `components/core/src/clp_s/OutputHandlerImpl.hpp/.cpp`, `components/core/src/clp_s/clp-s.cpp`) to identify logic/classes that were only needed to support the pre-refactor count/count-by-time aggregation flow and are now superseded or redundant.
- Audit the standalone reducer service module (`components/core/src/reducer/`, e.g. `CountOperator.cpp/.hpp`, `Pipeline.cpp/.hpp`, `Operator.cpp/.hpp`) to determine which pieces are still required by `clp_s` versus code that was only used by the old aggregation design.
- Remove or refactor any code identified as dead/obsolete, keeping only what's needed to support the current reducer output handler (count and count-by-time) and any future extensions.
- Update or add tests as needed to reflect the removal.
## Affected areas
- `components/core/src/clp_s/CommandLineArguments.cpp` / `.hpp`
- `components/core/src/clp_s/OutputHandlerImpl.cpp` / `.hpp`
- `components/core/src/clp_s/clp-s.cpp`
- `components/core/src/reducer/` (e.g. `CountOperator.cpp/.hpp`, `Pipeline.cpp/.hpp`, `Operator.cpp/.hpp`)
## Acceptance criteria
- Old/unused reducer-related code paths that are no longer exercised after the aggregation refactor are identified and removed.
- Existing reducer output handler functionality (count, count-by-time) continues to work as before.
- No dead code remains referencing the pre-refactor aggregation approach.
## References
- PR: https://github.com/y-scope/clp/pull/2345
- Comment: https://github.com/y-scope/clp/pull/2345#discussion_r3526356325
Requested by: `@davidlion`
EOF
)
issue_url=$(gh issue create \
--repo "$repo" \
--title "Remove obsolete reducer output handler code after aggregation refactor" \
--body "$body" \
--assignee davidlion)
echo "$issue_url"Length of output: 3103
@davidlion, done — I've created #2366 to track removing the old reducer-related code that's no longer needed after this aggregation refactor, assigned to you with backlinks to this PR and comment.
🧬 Code Graph Analysis Results
components/core/src/clp_s/CommandLineArguments.hpp
Relevant for reducer-related CLI behavior removal (it exposes how the reducer output handler is represented/parsed):
-
Lines 118-120
auto get_output_handler_options() const -> OutputHandlerOptionsVariant const& { return m_output_handler_options; }
-
Lines 188-192
parse_reducer_output_handler_options( boost::program_options::options_description const& options_description, std::vector<std::string> const& options, ReducerOutputHandlerOptions& reducer_options )
-
Lines 179-179
reject_aggregation_for_handler(std::string_view handler_name) const -> void
-
Lines 168-172
parse_aggregation_options( boost::program_options::variables_map const& parsed_options, int64_t count_by_time_bucket_size_ms, std::string_view aggregation_field ) -> std::optional<Aggregation>
🧠 Learnings used
Learnt from: SharafMohamed
Repo: y-scope/clp PR: 1313
File: components/core/tests/test-SchemaSearcher.cpp:136-137
Timestamp: 2025-12-09T01:57:04.839Z
Learning: In C++ files under the components directory, keep forward declarations even when a definition follows in the same scope (such as in anonymous namespaces) to maintain consistency across multiple methods. This style reduces churn and aligns with the existing convention in this repository for both source and tests.
Learnt from: ShangDanLuXian
Repo: y-scope/clp PR: 2010
File: components/core/src/clp_s/filter/FilterBuilder.cpp:24-33
Timestamp: 2026-02-20T15:19:43.528Z
Learning: In the components/core/src/clp_s module, FileWriter usage across multipleCpp files (e.g., JsonConstructor.cpp, SchemaMap.cpp, FilterBuilder.cpp) intentionally omits try-catch around write operations, and treats potential write failures as acceptable data loss. Do not add per-write try-catch in these files; instead, rely on the caller/integration layer to handle exceptions. Apply this pattern to similar FileWriter usage within this module, ensuring reviewers understand that error handling is deferred and that this behavior is consistent across related files.
Learnt from: Bill-hbrhbr
Repo: y-scope/clp PR: 2144
File: components/core/src/clp_s/search/ast/SearchUtils.cpp:10-11
Timestamp: 2026-03-29T20:05:44.816Z
Learning: In y-scope/clp, the include style is intentional within `components/core/src/clp_s/**`: use angle-bracket includes (e.g., `<clp_s/archive_constants.hpp>`) for headers that belong to external CMake targets, and use quoted relative includes (e.g., `"ConvertToExists.hpp"`) for headers within the same CMake target. Do not flag angle vs. quote include style as an inconsistency when it follows this rule.
| #include "../clp/type_utils.hpp" | ||
| #include "../reducer/types.hpp" | ||
| #include "FileReader.hpp" | ||
| #include "search/ast/SearchUtils.hpp" |
Member
There was a problem hiding this comment.
This was marked resolved, but the change was never applied...
|
|
||
| namespace clp_s { | ||
| /** | ||
| * A single typed value in an aggregation result document. |
Member
There was a problem hiding this comment.
I think with the way Aggregation is used in other places it should be possessive here (you also wrote it this way in another place too).
Suggested change
| * A single typed value in an aggregation result document. | |
| * A single typed value in an aggregation's result document. |
| using AggregationValue = std::variant<int64_t, double, std::string>; | ||
|
|
||
| /** | ||
| * One aggregation result document: an ordered list of typed key-value pairs. |
Member
There was a problem hiding this comment.
Suggested change
| * One aggregation result document: an ordered list of typed key-value pairs. | |
| * One aggregation's result document: an ordered list of typed key-value pairs. |
Comment on lines
+56
to
+57
| // The tokenizer strips the namespace prefix out of the path, but namespaced fields live | ||
| // under a top-level object keyed by that namespace. Prepend it back so the path matches. |
Member
There was a problem hiding this comment.
Suggested change
| // The tokenizer strips the namespace prefix out of the path, but namespaced fields live | |
| // under a top-level object keyed by that namespace. Prepend it back so the path matches. |
I think this counts as an implementation detail, so it goes above the function definition.
- Rename the per-record aggregator classes and their variant from `*Aggregation`/`Aggregation` to `*Aggregator`/`Aggregator`, keeping "aggregation" for the operation's data/output (value, result, sink, output handler). - Rename ambiguous `_ms` identifiers to `_millisecs`. - Use `<clp_s/...>` includes, brace-init constexpr members, and tidy doc-comments (possessive wording, hoist implementation notes). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
davidlion
requested changes
Jul 9, 2026
davidlion
left a comment
davidlion
left a comment
Member
There was a problem hiding this comment.
very small changes and lgtm
davidlion
reviewed
Jul 9, 2026
Member
|
I didn't think of this for the previous PRs, but what do you think of the title: |
davemarco
changed the title
… category comments.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
Adds --min and --max search aggregations, computing the minimum/maximum value of a numeric field and writing the result to stdout or the results cache. Continues the aggregation work from #2326 and #2342.
Outputs one document per archive, for both destinations:
{archive_id, field, min|max}
Changes
Checklist
breaking change.
Validation performed
Validated a few queries on mongoDb dataset
Summary by CodeRabbit