Repository navigation
Add COLLECT reducer to FT.AGGREGATE #524
Next
Next commit
Add COLLECT reducer to FT.AGGREGATE
Introduce the COLLECT reducer for FT.AGGREGATE GROUPBY, which gathers per-document projections within each group and returns them as an array of per-entry maps under the reducer alias, optionally sorted and bounded. - Add CollectReducer with Fields(...)/FieldsAll(), SortBy(...)/SortByAsc/ SortByDesc, Limit(...) and As(...), plus the Reducers.Collect() factory. Field and sort names are normalized to a single '@' prefix on the wire, and <narg> covers the FIELDS/SORTBY/LIMIT tokens (excluding AS <alias>), matching the RediSearch grammar. Nested COLLECT columns are read through the existing AggregationResult parsing across RESP2 and RESP3. - Register the new public API symbols in PublicAPI.Unshipped.txt. COLLECT is experimental and gated behind search-enable-unstable-features on the server.
- Loading branch information
commit db6f4901e960f1ead33853739892ec85a91bdc2a
There are no files selected for viewing
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
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,153 @@ | ||
| using NRedisStack.Search.Literals; | ||
|
|
||
| namespace NRedisStack.Search.Aggregation; | ||
|
|
||
| /// <summary> | ||
| /// Reducer for <c>REDUCE COLLECT</c>, which gathers per-document projections within a <c>GROUPBY</c> group and returns them | ||
| /// as an array of per-entry maps under the reducer alias, optionally sorted and bounded. | ||
| /// <para> | ||
| /// The grammar implemented by this reducer is: | ||
| /// <code> | ||
| /// REDUCE COLLECT <narg> | ||
| /// FIELDS ( * | <num_fields> <@field> [<@field> ...] ) | ||
| /// [SORTBY <narg> <@field> [ASC|DESC] [<@field> [ASC|DESC] ...]] | ||
| /// [LIMIT <offset> <count>] | ||
| /// [AS <alias>] | ||
| /// </code> | ||
| /// </para> | ||
| /// <para> | ||
| /// Field and sort-key names are referenced with an <c>@</c> prefix on the wire (this builder adds it automatically when it is | ||
| /// missing). The output map keys are the bare names. <c>FIELDS *</c> projects whatever the pipeline has materialized at the | ||
| /// <c>COLLECT</c> stage; it does not implicitly fetch the full document. | ||
| /// </para> | ||
| /// <para> | ||
| /// <b>Experimental.</b> Both the underlying Redis Search feature and this API may change. <c>COLLECT</c> is gated behind | ||
| /// <c>search-enable-unstable-features</c>; enable it on the server (for example via | ||
| /// <c>CONFIG SET search-enable-unstable-features yes</c>) before issuing aggregations that use this reducer, otherwise the | ||
| /// server replies with an error. | ||
| /// </para> | ||
| /// </summary> | ||
| /// <seealso cref="Reducers.Collect()"/> | ||
| public sealed class CollectReducer : Reducer | ||
| { | ||
| private bool _allFields = false; | ||
| private readonly List<string> _fields = new List<string>(); | ||
| private readonly List<SortedField> _sortFields = new List<SortedField>(); | ||
| private int? _limitOffset; | ||
| private int? _limitCount; | ||
|
|
||
| internal CollectReducer() : base(null) { } | ||
|
|
||
| public override string Name => "COLLECT"; | ||
|
|
||
| /// <summary> | ||
| /// Project the named fields for every document in the group. Names may be supplied with or without a leading <c>@</c>; | ||
| /// the builder normalizes each to a single <c>@<name></c> on the wire. Use <c>@__key</c> or ordinary document field | ||
| /// names. Mutually exclusive with <see cref="FieldsAll"/>; may be called multiple times to append further fields. | ||
| /// </summary> | ||
| public CollectReducer Fields(params string[] fields) | ||
| { | ||
| if (_allFields) | ||
| throw new InvalidOperationException("REDUCE COLLECT cannot mix FIELDS * with explicit field names"); | ||
| _fields.AddRange(fields); | ||
| return this; | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Project every field present in the pipeline at the <c>COLLECT</c> stage (<c>FIELDS *</c>). Per the COLLECT | ||
| /// specification, <c>*</c> does not trigger an implicit load — fields must already be in the pipeline (typically via | ||
| /// <c>LOAD *</c> or because they are grouping keys / reducer aliases). Mutually exclusive with <see cref="Fields"/>. | ||
| /// </summary> | ||
| public CollectReducer FieldsAll() | ||
| { | ||
| if (_fields.Count > 0) | ||
| throw new InvalidOperationException("REDUCE COLLECT cannot mix FIELDS * with explicit field names"); | ||
| _allFields = true; | ||
| return this; | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// In-group sort by one or more fields. May be called multiple times to append further sort keys. | ||
| /// </summary> | ||
| public CollectReducer SortBy(params SortedField[] fields) | ||
| { | ||
| _sortFields.AddRange(fields); | ||
| return this; | ||
| } | ||
|
|
||
| /// <summary>Convenience for <c>SortBy(SortedField.Asc(field))</c>.</summary> | ||
| public CollectReducer SortByAsc(string field) | ||
| { | ||
| _sortFields.Add(SortedField.Asc(field)); | ||
| return this; | ||
| } | ||
|
|
||
| /// <summary>Convenience for <c>SortBy(SortedField.Desc(field))</c>.</summary> | ||
| public CollectReducer SortByDesc(string field) | ||
| { | ||
| _sortFields.Add(SortedField.Desc(field)); | ||
| return this; | ||
| } | ||
|
|
||
| /// <summary>Bound the output per group to the first <paramref name="count"/> entries (offset 0).</summary> | ||
| public CollectReducer Limit(int count) => Limit(0, count); | ||
|
|
||
| /// <summary>Bound the output per group to <paramref name="count"/> entries starting at <paramref name="offset"/>.</summary> | ||
| public CollectReducer Limit(int offset, int count) | ||
| { | ||
| if (offset < 0 || count < 0) | ||
| throw new ArgumentException("LIMIT offset and count must be non-negative"); | ||
| _limitOffset = offset; | ||
| _limitCount = count; | ||
| return this; | ||
| } | ||
|
|
||
| protected override int GetOwnArgsCount() | ||
| { | ||
| if (!_allFields && _fields.Count == 0) | ||
| throw new InvalidOperationException( | ||
| "REDUCE COLLECT requires either Fields(...) or FieldsAll() to be configured"); | ||
|
|
||
| int count = _allFields ? 2 : 2 + _fields.Count; | ||
| if (_sortFields.Count > 0) | ||
| count += 2 + _sortFields.Count * 2; // SORTBY <narg> then a @field/ASC|DESC pair per key | ||
| if (_limitOffset != null) | ||
| count += 3; // LIMIT <offset> <count> | ||
| return count; | ||
| } | ||
|
|
||
| protected override void AddOwnArgs(List<object> args) | ||
| { | ||
| args.Add(SearchArgs.FIELDS); | ||
| if (_allFields) | ||
| { | ||
| args.Add("*"); | ||
| } | ||
| else | ||
| { | ||
| args.Add(_fields.Count); | ||
| foreach (var field in _fields) | ||
| args.Add(WithAtPrefix(field)); | ||
| } | ||
|
|
||
| if (_sortFields.Count > 0) | ||
| { | ||
| args.Add(SearchArgs.SORTBY); | ||
| args.Add(_sortFields.Count * 2); | ||
| foreach (var sortField in _sortFields) | ||
| { | ||
| args.Add(WithAtPrefix(sortField.FieldName)); | ||
| args.Add(sortField.Order.ToString()); | ||
| } | ||
| } | ||
|
|
||
| if (_limitOffset != null) | ||
| { | ||
| args.Add(SearchArgs.LIMIT); | ||
| args.Add(_limitOffset.Value); | ||
| args.Add(_limitCount!.Value); | ||
| } | ||
| } | ||
|
|
||
| private static string WithAtPrefix(string name) => name.StartsWith("@") ? name : "@" + name; | ||
| } | ||
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
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,90 @@ | ||
| using NRedisStack.Search; | ||
| using NRedisStack.Search.Aggregation; | ||
| using Xunit; | ||
|
|
||
| namespace NRedisStack.Tests.Search; | ||
|
|
||
| /// <summary> | ||
| /// Unit tests for <see cref="CollectReducer"/> — verify that the builder produces the exact token layout that the Redis | ||
| /// Search COLLECT grammar expects. These tests do not require a running server. | ||
| /// </summary> | ||
| public class CollectReducerTests | ||
| { | ||
| [Fact] | ||
| public void CollectExplicitFieldsSortByLimitAndAlias() | ||
| { | ||
| var request = new AggregationRequest("*").GroupBy("@color", Reducers.Collect() | ||
| .Fields("fruit", "sweetness") | ||
| .SortBy(SortedField.Desc("sweetness")) | ||
| .Limit(0, 2) | ||
| .As("top")); | ||
|
|
||
| // narg = FIELDS 2 @fruit @sweetness SORTBY 2 @sweetness DESC LIMIT 0 2 = 11 | ||
| object[] expected = | ||
| [ | ||
| "*", "GROUPBY", 1, "@color", | ||
| "REDUCE", "COLLECT", 11, | ||
| "FIELDS", 2, "@fruit", "@sweetness", | ||
| "SORTBY", 2, "@sweetness", "DESC", | ||
| "LIMIT", 0, 2, | ||
| "AS", "top" | ||
| ]; | ||
|
|
||
| Assert.Equal(expected, request.GetArgs()); | ||
| } | ||
|
|
||
| [Fact] | ||
| public void CollectFieldsAllSortByDescAndLimitWithoutAlias() | ||
| { | ||
| var request = new AggregationRequest("*").GroupBy("@color", Reducers.Collect() | ||
| .FieldsAll() | ||
| .SortByDesc("sweetness") | ||
| .Limit(2)); | ||
|
|
||
| // narg = FIELDS * SORTBY 2 @sweetness DESC LIMIT 0 2 = 9 | ||
| object[] expected = | ||
| [ | ||
| "*", "GROUPBY", 1, "@color", | ||
| "REDUCE", "COLLECT", 9, | ||
| "FIELDS", "*", | ||
| "SORTBY", 2, "@sweetness", "DESC", | ||
| "LIMIT", 0, 2 | ||
| ]; | ||
|
|
||
| Assert.Equal(expected, request.GetArgs()); | ||
| } | ||
|
|
||
| [Fact] | ||
| public void CollectMultipleSortKeysNormalizeAtPrefix() | ||
| { | ||
| var request = new AggregationRequest("*").GroupBy("@color", Reducers.Collect() | ||
| .Fields("@__key", "fruit") | ||
| .SortBy(SortedField.Desc("@sweetness"), SortedField.Asc("__key")) | ||
| .As("top")); | ||
|
|
||
| // narg = FIELDS 2 @__key @fruit SORTBY 4 @sweetness DESC @__key ASC = 10 | ||
| object[] expected = | ||
| [ | ||
| "*", "GROUPBY", 1, "@color", | ||
| "REDUCE", "COLLECT", 10, | ||
| "FIELDS", 2, "@__key", "@fruit", | ||
| "SORTBY", 4, "@sweetness", "DESC", "@__key", "ASC", | ||
| "AS", "top" | ||
| ]; | ||
|
|
||
| Assert.Equal(expected, request.GetArgs()); | ||
| } | ||
|
|
||
| [Fact] | ||
| public void CollectRejectsInvalidLocalUsage() | ||
| { | ||
| Assert.Throws<InvalidOperationException>(() => Reducers.Collect().Fields("a").FieldsAll()); | ||
| Assert.Throws<InvalidOperationException>(() => Reducers.Collect().FieldsAll().Fields("a")); | ||
| Assert.Throws<ArgumentException>(() => Reducers.Collect().Limit(-1, 5)); | ||
|
|
||
| // Serialization (which happens eagerly when the reducer is attached to a GROUPBY) fails when neither | ||
| // Fields(...) nor FieldsAll() was configured. | ||
| Assert.Throws<InvalidOperationException>( | ||
| () => new AggregationRequest("*").GroupBy("@color", Reducers.Collect().As("top"))); | ||
| } | ||
| } |
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
Oops, something went wrong.
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.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Experimental COLLECT API missing
[Experimental]attributeMedium Severity
The
CollectReducerclass andReducers.Collect()factory are described as experimental in both the PR description and the XML doc comments, yet they are not marked with the[Experimental]attribute. The codebase has established infrastructure for this inExperiments.cs(with retired IDs NRS001 and NRS002 and correspondingdocs/exp/pages). Without the attribute, consumers get no compiler warning when depending on an unstable API that may change or break.Additional Locations (1)
src/NRedisStack/Search/Reducers.cs#L94-L95Reviewed by Cursor Bugbot for commit 92e8220. Configure here.