From db6f4901e960f1ead33853739892ec85a91bdc2a Mon Sep 17 00:00:00 2001 From: Igor Malinovskiy Date: Thu, 16 Jul 2026 19:14:46 +0200 Subject: [PATCH 1/2] 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 covers the FIELDS/SORTBY/LIMIT tokens (excluding AS ), 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. --- .../PublicAPI/PublicAPI.Unshipped.txt | 10 ++ src/NRedisStack/Search/CollectReducer.cs | 153 ++++++++++++++++++ src/NRedisStack/Search/Reducers.cs | 16 ++ .../Search/CollectReducerTests.cs | 90 +++++++++++ tests/NRedisStack.Tests/Search/SearchTests.cs | 75 +++++++++ 5 files changed, 344 insertions(+) create mode 100644 src/NRedisStack/Search/CollectReducer.cs create mode 100644 tests/NRedisStack.Tests/Search/CollectReducerTests.cs diff --git a/src/NRedisStack/PublicAPI/PublicAPI.Unshipped.txt b/src/NRedisStack/PublicAPI/PublicAPI.Unshipped.txt index 465a8ecf..70eb2976 100644 --- a/src/NRedisStack/PublicAPI/PublicAPI.Unshipped.txt +++ b/src/NRedisStack/PublicAPI/PublicAPI.Unshipped.txt @@ -47,3 +47,13 @@ NRedisStack.Search.SearchResult.Warnings.get -> string![]! NRedisStack.Search.AggregationResult.Warnings.get -> string![]! NRedisStack.Search.HybridSearchResult.Warnings.get -> string![]! NRedisStack.Search.AggregationRequest.AddScores() -> NRedisStack.Search.AggregationRequest! +NRedisStack.Search.Aggregation.CollectReducer +override NRedisStack.Search.Aggregation.CollectReducer.Name.get -> string! +NRedisStack.Search.Aggregation.CollectReducer.Fields(params string![]! fields) -> NRedisStack.Search.Aggregation.CollectReducer! +NRedisStack.Search.Aggregation.CollectReducer.FieldsAll() -> NRedisStack.Search.Aggregation.CollectReducer! +NRedisStack.Search.Aggregation.CollectReducer.SortBy(params NRedisStack.Search.Aggregation.SortedField![]! fields) -> NRedisStack.Search.Aggregation.CollectReducer! +NRedisStack.Search.Aggregation.CollectReducer.SortByAsc(string! field) -> NRedisStack.Search.Aggregation.CollectReducer! +NRedisStack.Search.Aggregation.CollectReducer.SortByDesc(string! field) -> NRedisStack.Search.Aggregation.CollectReducer! +NRedisStack.Search.Aggregation.CollectReducer.Limit(int count) -> NRedisStack.Search.Aggregation.CollectReducer! +NRedisStack.Search.Aggregation.CollectReducer.Limit(int offset, int count) -> NRedisStack.Search.Aggregation.CollectReducer! +static NRedisStack.Search.Aggregation.Reducers.Collect() -> NRedisStack.Search.Aggregation.CollectReducer! \ No newline at end of file diff --git a/src/NRedisStack/Search/CollectReducer.cs b/src/NRedisStack/Search/CollectReducer.cs new file mode 100644 index 00000000..a880e114 --- /dev/null +++ b/src/NRedisStack/Search/CollectReducer.cs @@ -0,0 +1,153 @@ +using NRedisStack.Search.Literals; + +namespace NRedisStack.Search.Aggregation; + +/// +/// Reducer for REDUCE COLLECT, which gathers per-document projections within a GROUPBY group and returns them +/// as an array of per-entry maps under the reducer alias, optionally sorted and bounded. +/// +/// The grammar implemented by this reducer is: +/// +/// REDUCE COLLECT <narg> +/// FIELDS ( * | <num_fields> <@field> [<@field> ...] ) +/// [SORTBY <narg> <@field> [ASC|DESC] [<@field> [ASC|DESC] ...]] +/// [LIMIT <offset> <count>] +/// [AS <alias>] +/// +/// +/// +/// Field and sort-key names are referenced with an @ prefix on the wire (this builder adds it automatically when it is +/// missing). The output map keys are the bare names. FIELDS * projects whatever the pipeline has materialized at the +/// COLLECT stage; it does not implicitly fetch the full document. +/// +/// +/// Experimental. Both the underlying Redis Search feature and this API may change. COLLECT is gated behind +/// search-enable-unstable-features; enable it on the server (for example via +/// CONFIG SET search-enable-unstable-features yes) before issuing aggregations that use this reducer, otherwise the +/// server replies with an error. +/// +/// +/// +public sealed class CollectReducer : Reducer +{ + private bool _allFields = false; + private readonly List _fields = new List(); + private readonly List _sortFields = new List(); + private int? _limitOffset; + private int? _limitCount; + + internal CollectReducer() : base(null) { } + + public override string Name => "COLLECT"; + + /// + /// Project the named fields for every document in the group. Names may be supplied with or without a leading @; + /// the builder normalizes each to a single @<name> on the wire. Use @__key or ordinary document field + /// names. Mutually exclusive with ; may be called multiple times to append further fields. + /// + 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; + } + + /// + /// Project every field present in the pipeline at the COLLECT stage (FIELDS *). Per the COLLECT + /// specification, * does not trigger an implicit load — fields must already be in the pipeline (typically via + /// LOAD * or because they are grouping keys / reducer aliases). Mutually exclusive with . + /// + public CollectReducer FieldsAll() + { + if (_fields.Count > 0) + throw new InvalidOperationException("REDUCE COLLECT cannot mix FIELDS * with explicit field names"); + _allFields = true; + return this; + } + + /// + /// In-group sort by one or more fields. May be called multiple times to append further sort keys. + /// + public CollectReducer SortBy(params SortedField[] fields) + { + _sortFields.AddRange(fields); + return this; + } + + /// Convenience for SortBy(SortedField.Asc(field)). + public CollectReducer SortByAsc(string field) + { + _sortFields.Add(SortedField.Asc(field)); + return this; + } + + /// Convenience for SortBy(SortedField.Desc(field)). + public CollectReducer SortByDesc(string field) + { + _sortFields.Add(SortedField.Desc(field)); + return this; + } + + /// Bound the output per group to the first entries (offset 0). + public CollectReducer Limit(int count) => Limit(0, count); + + /// Bound the output per group to entries starting at . + 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 then a @field/ASC|DESC pair per key + if (_limitOffset != null) + count += 3; // LIMIT + return count; + } + + protected override void AddOwnArgs(List 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; +} diff --git a/src/NRedisStack/Search/Reducers.cs b/src/NRedisStack/Search/Reducers.cs index c595b8e5..d8a20e23 100644 --- a/src/NRedisStack/Search/Reducers.cs +++ b/src/NRedisStack/Search/Reducers.cs @@ -78,6 +78,22 @@ protected override void AddOwnArgs(List args) public static Reducer ToList(string field) => new SingleFieldReducer("TOLIST", field); + /// + /// REDUCE COLLECT — gather per-document projections within a GROUPBY group and return them as an array of per-entry maps + /// under the reducer alias, optionally sorted and bounded. + /// + /// Configure the projected fields via or , then + /// optionally chain and before calling + /// . + /// + /// + /// Experimental. Both the underlying Redis Search feature and this API may change. Before issuing COLLECT queries the + /// server must be configured with CONFIG SET search-enable-unstable-features yes. + /// + /// + /// + public static CollectReducer Collect() => new CollectReducer(); + public static Reducer RandomSample(string field, int size) => new RandomSampleReducer(field, size); private sealed class RandomSampleReducer : Reducer diff --git a/tests/NRedisStack.Tests/Search/CollectReducerTests.cs b/tests/NRedisStack.Tests/Search/CollectReducerTests.cs new file mode 100644 index 00000000..6a406ec2 --- /dev/null +++ b/tests/NRedisStack.Tests/Search/CollectReducerTests.cs @@ -0,0 +1,90 @@ +using NRedisStack.Search; +using NRedisStack.Search.Aggregation; +using Xunit; + +namespace NRedisStack.Tests.Search; + +/// +/// Unit tests for — verify that the builder produces the exact token layout that the Redis +/// Search COLLECT grammar expects. These tests do not require a running server. +/// +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(() => Reducers.Collect().Fields("a").FieldsAll()); + Assert.Throws(() => Reducers.Collect().FieldsAll().Fields("a")); + Assert.Throws(() => 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( + () => new AggregationRequest("*").GroupBy("@color", Reducers.Collect().As("top"))); + } +} diff --git a/tests/NRedisStack.Tests/Search/SearchTests.cs b/tests/NRedisStack.Tests/Search/SearchTests.cs index 2ce2ddbf..09ed3693 100644 --- a/tests/NRedisStack.Tests/Search/SearchTests.cs +++ b/tests/NRedisStack.Tests/Search/SearchTests.cs @@ -240,6 +240,81 @@ public void TestAggregations(string endpointId) Assert.Equal(10, r2.GetLong("sum")); } + [SkipIfRedisTheory(Comparison.LessThan, "8.7.0")] + [MemberData(nameof(EndpointsFixture.Env.AllEnvironments), MemberType = typeof(EndpointsFixture.Env))] + public void TestAggregationCollectReducer(string endpointId) + { + SkipClusterPre8(endpointId); + var redis = GetConnection(endpointId); + IDatabase db = GetCleanDatabase(endpointId); + var ft = db.FT(); + + // COLLECT is gated behind unstable features. In a cluster the coordinator fans the query out to every shard, + // so the flag must be enabled on all nodes rather than only the one this connection happens to route to. + try + { + foreach (var endPoint in redis.GetEndPoints()) + { + var server = redis.GetServer(endPoint); + if (!server.IsConnected) continue; + server.Execute("CONFIG", "SET", "search-enable-unstable-features", "yes"); + } + } + catch (RedisException ex) + { + Assert.Skip($"search-enable-unstable-features is not configurable on this Redis build: {ex.Message}"); + } + + Schema sc = new(); + sc.AddTagField("color"); + sc.AddTagField("fruit"); + sc.AddNumericField("sweetness", true); + ft.Create(index, FTCreateParams.CreateParams(), sc); + + AddDocument(db, new Document("fruit:1").Set("color", "yellow").Set("fruit", "apple").Set("sweetness", 6)); + AddDocument(db, new Document("fruit:2").Set("color", "yellow").Set("fruit", "banana").Set("sweetness", 5)); + AddDocument(db, new Document("fruit:3").Set("color", "yellow").Set("fruit", "lemon").Set("sweetness", 2)); + AddDocument(db, new Document("fruit:4").Set("color", "red").Set("fruit", "cherry").Set("sweetness", 7)); + + var request = new AggregationRequest("*").GroupBy("@color", Reducers.Collect() + .Fields("fruit", "sweetness") + .SortBy(SortedField.Desc("sweetness")) + .Limit(0, 2) + .As("top")); + + AggregationResult res; + try + { + res = ft.Aggregate(index, request); + } + catch (RedisException ex) + { + Assert.Skip($"FT.AGGREGATE REDUCE COLLECT not supported by this Redis Search build: {ex.Message}"); + return; + } + + object? yellowTop = null; + for (int i = 0; i < res.TotalResults; i++) + { + var row = res.GetRow(i); + if (row.GetString("color") == "yellow") + { + yellowTop = row.Get("top"); + } + } + + Assert.NotNull(yellowTop); + Assert.IsAssignableFrom(yellowTop); + int entryCount = 0; + foreach (var _ in (System.Collections.IEnumerable)yellowTop!) + { + entryCount++; + } + + // LIMIT 0 2 caps the yellow group at 2 entries; SORTBY @sweetness DESC keeps apple (6) and banana (5). + Assert.Equal(2, entryCount); + } + [Theory] [MemberData(nameof(EndpointsFixture.Env.AllEnvironments), MemberType = typeof(EndpointsFixture.Env))] public async Task TestAggregationsAsync(string endpointId) From b630fd1ae00fd356ccbcd3d6e4b911177c941b0c Mon Sep 17 00:00:00 2001 From: Igor Malinovskiy Date: Fri, 17 Jul 2026 11:12:08 +0200 Subject: [PATCH 2/2] Clean up --- src/NRedisStack/Search/CollectReducer.cs | 34 +++++++++++++------ .../Search/CollectReducerTests.cs | 22 ++++++++++++ tests/NRedisStack.Tests/Search/SearchTests.cs | 11 +----- 3 files changed, 47 insertions(+), 20 deletions(-) diff --git a/src/NRedisStack/Search/CollectReducer.cs b/src/NRedisStack/Search/CollectReducer.cs index a880e114..878697cc 100644 --- a/src/NRedisStack/Search/CollectReducer.cs +++ b/src/NRedisStack/Search/CollectReducer.cs @@ -21,6 +21,11 @@ namespace NRedisStack.Search.Aggregation; /// COLLECT stage; it does not implicitly fetch the full document. /// /// +/// Configure the reducer fully before attaching it to a GROUPBY: +/// serializes the reducer immediately, so builder calls made after that point cannot reach the wire and throw +/// . +/// +/// /// Experimental. Both the underlying Redis Search feature and this API may change. COLLECT is gated behind /// search-enable-unstable-features; enable it on the server (for example via /// CONFIG SET search-enable-unstable-features yes) before issuing aggregations that use this reducer, otherwise the @@ -35,6 +40,7 @@ public sealed class CollectReducer : Reducer private readonly List _sortFields = new List(); private int? _limitOffset; private int? _limitCount; + private bool _serialized; internal CollectReducer() : base(null) { } @@ -47,6 +53,7 @@ internal CollectReducer() : base(null) { } /// public CollectReducer Fields(params string[] fields) { + EnsureMutable(); if (_allFields) throw new InvalidOperationException("REDUCE COLLECT cannot mix FIELDS * with explicit field names"); _fields.AddRange(fields); @@ -60,6 +67,7 @@ public CollectReducer Fields(params string[] fields) /// public CollectReducer FieldsAll() { + EnsureMutable(); if (_fields.Count > 0) throw new InvalidOperationException("REDUCE COLLECT cannot mix FIELDS * with explicit field names"); _allFields = true; @@ -71,23 +79,16 @@ public CollectReducer FieldsAll() /// public CollectReducer SortBy(params SortedField[] fields) { + EnsureMutable(); _sortFields.AddRange(fields); return this; } /// Convenience for SortBy(SortedField.Asc(field)). - public CollectReducer SortByAsc(string field) - { - _sortFields.Add(SortedField.Asc(field)); - return this; - } + public CollectReducer SortByAsc(string field) => SortBy(SortedField.Asc(field)); /// Convenience for SortBy(SortedField.Desc(field)). - public CollectReducer SortByDesc(string field) - { - _sortFields.Add(SortedField.Desc(field)); - return this; - } + public CollectReducer SortByDesc(string field) => SortBy(SortedField.Desc(field)); /// Bound the output per group to the first entries (offset 0). public CollectReducer Limit(int count) => Limit(0, count); @@ -95,6 +96,7 @@ public CollectReducer SortByDesc(string field) /// Bound the output per group to entries starting at . public CollectReducer Limit(int offset, int count) { + EnsureMutable(); if (offset < 0 || count < 0) throw new ArgumentException("LIMIT offset and count must be non-negative"); _limitOffset = offset; @@ -118,6 +120,10 @@ protected override int GetOwnArgsCount() protected override void AddOwnArgs(List args) { + // GroupBy serializes the reducer eagerly; once the tokens are emitted, later builder + // calls could never reach the wire, so reject them instead of silently ignoring them. + _serialized = true; + args.Add(SearchArgs.FIELDS); if (_allFields) { @@ -149,5 +155,13 @@ protected override void AddOwnArgs(List args) } } + private void EnsureMutable() + { + if (_serialized) + throw new InvalidOperationException( + "REDUCE COLLECT cannot be modified after it has been serialized into an aggregation request; " + + "configure Fields/SortBy/Limit before passing the reducer to GroupBy"); + } + private static string WithAtPrefix(string name) => name.StartsWith("@") ? name : "@" + name; } diff --git a/tests/NRedisStack.Tests/Search/CollectReducerTests.cs b/tests/NRedisStack.Tests/Search/CollectReducerTests.cs index 6a406ec2..2d60f699 100644 --- a/tests/NRedisStack.Tests/Search/CollectReducerTests.cs +++ b/tests/NRedisStack.Tests/Search/CollectReducerTests.cs @@ -87,4 +87,26 @@ public void CollectRejectsInvalidLocalUsage() Assert.Throws( () => new AggregationRequest("*").GroupBy("@color", Reducers.Collect().As("top"))); } + + [Fact] + public void CollectRejectsMutationAfterAttachingToGroupBy() + { + // GroupBy serializes the reducer eagerly, so builder calls made afterwards could never + // reach the wire; they must throw rather than be silently dropped. + var collect = Reducers.Collect().Fields("fruit"); + var request = new AggregationRequest("*").GroupBy("@color", collect); + + Assert.Throws(() => collect.Fields("sweetness")); + Assert.Throws(() => collect.SortByDesc("sweetness")); + Assert.Throws(() => collect.Limit(0, 2)); + + // The serialized request is unchanged by the rejected calls. + object[] expected = + [ + "*", "GROUPBY", 1, "@color", + "REDUCE", "COLLECT", 3, + "FIELDS", 1, "@fruit" + ]; + Assert.Equal(expected, request.GetArgs()); + } } diff --git a/tests/NRedisStack.Tests/Search/SearchTests.cs b/tests/NRedisStack.Tests/Search/SearchTests.cs index 09ed3693..7a1ad54f 100644 --- a/tests/NRedisStack.Tests/Search/SearchTests.cs +++ b/tests/NRedisStack.Tests/Search/SearchTests.cs @@ -282,16 +282,7 @@ public void TestAggregationCollectReducer(string endpointId) .Limit(0, 2) .As("top")); - AggregationResult res; - try - { - res = ft.Aggregate(index, request); - } - catch (RedisException ex) - { - Assert.Skip($"FT.AGGREGATE REDUCE COLLECT not supported by this Redis Search build: {ex.Message}"); - return; - } + AggregationResult res = ft.Aggregate(index, request); object? yellowTop = null; for (int i = 0; i < res.TotalResults; i++)