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..878697cc --- /dev/null +++ b/src/NRedisStack/Search/CollectReducer.cs @@ -0,0 +1,167 @@ +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. +/// +/// +/// 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 +/// 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; + private bool _serialized; + + 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) + { + EnsureMutable(); + 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() + { + EnsureMutable(); + 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) + { + EnsureMutable(); + _sortFields.AddRange(fields); + return this; + } + + /// Convenience for SortBy(SortedField.Asc(field)). + public CollectReducer SortByAsc(string field) => SortBy(SortedField.Asc(field)); + + /// Convenience for SortBy(SortedField.Desc(field)). + 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); + + /// 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; + _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) + { + // 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) + { + 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 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/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..2d60f699 --- /dev/null +++ b/tests/NRedisStack.Tests/Search/CollectReducerTests.cs @@ -0,0 +1,112 @@ +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"))); + } + + [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 2ce2ddbf..7a1ad54f 100644 --- a/tests/NRedisStack.Tests/Search/SearchTests.cs +++ b/tests/NRedisStack.Tests/Search/SearchTests.cs @@ -240,6 +240,72 @@ 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 = ft.Aggregate(index, request); + + 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)