Add COLLECT reducer to FT.AGGREGATE

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!

src/NRedisStack/Search/CollectReducer.cs

@@ -0,0 +1,167 @@
+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 &lt;narg&gt;
+///     FIELDS ( * | &lt;num_fields&gt; &lt;@field&gt; [&lt;@field&gt; ...] )
+///     [SORTBY &lt;narg&gt; &lt;@field&gt; [ASC|DESC] [&lt;@field&gt; [ASC|DESC] ...]]
+///     [LIMIT &lt;offset&gt; &lt;count&gt;]
+///   [AS &lt;alias&gt;]
+/// </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>
+/// Configure the reducer fully before attaching it to a <c>GROUPBY</c>: <see cref="AggregationRequest.GroupBy(string, Reducer[])"/>
+/// serializes the reducer immediately, so builder calls made after that point cannot reach the wire and throw
+/// <see cref="InvalidOperationException"/>.
+/// </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;
+    private bool _serialized;
+
+    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>@&lt;name&gt;</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)
+    {
+        EnsureMutable();
+        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()
+    {
+        EnsureMutable();
+        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)
+    {
+        EnsureMutable();
+        _sortFields.AddRange(fields);
+        return this;
+    }
+
+    /// <summary>Convenience for <c>SortBy(SortedField.Asc(field))</c>.</summary>
+    public CollectReducer SortByAsc(string field) => SortBy(SortedField.Asc(field));
+
+    /// <summary>Convenience for <c>SortBy(SortedField.Desc(field))</c>.</summary>
+    public CollectReducer SortByDesc(string field) => SortBy(SortedField.Desc(field));
+
+    /// <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)
+    {
+        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 <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)
+    {
+        // 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;
+}

src/NRedisStack/Search/Reducers.cs

@@ -78,6 +78,22 @@ protected override void AddOwnArgs(List<object> args)
 
     public static Reducer ToList(string field) => new SingleFieldReducer("TOLIST", field);
 
+    /// <summary>
+    /// 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.
+    /// <para>
+    /// Configure the projected fields via <see cref="CollectReducer.Fields"/> or <see cref="CollectReducer.FieldsAll"/>, then
+    /// optionally chain <see cref="CollectReducer.SortBy"/> and <see cref="CollectReducer.Limit(int, int)"/> before calling
+    /// <see cref="Reducer.As(string)"/>.
+    /// </para>
+    /// <para>
+    /// <b>Experimental.</b> Both the underlying Redis Search feature and this API may change. Before issuing COLLECT queries the
+    /// server must be configured with <c>CONFIG SET search-enable-unstable-features yes</c>.
+    /// </para>
+    /// </summary>
+    /// <seealso cref="CollectReducer"/>
+    public static CollectReducer Collect() => new CollectReducer();
+
     public static Reducer RandomSample(string field, int size) => new RandomSampleReducer(field, size);
 
     private sealed class RandomSampleReducer : Reducer

tests/NRedisStack.Tests/Search/CollectReducerTests.cs

@@ -0,0 +1,112 @@
+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")));
+    }
+
+    [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<InvalidOperationException>(() => collect.Fields("sweetness"));
+        Assert.Throws<InvalidOperationException>(() => collect.SortByDesc("sweetness"));
+        Assert.Throws<InvalidOperationException>(() => 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());
+    }
+}

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<System.Collections.IEnumerable>(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)