feat(search): add COLLECT reducer for FT.AGGREGATE (#3886)

* feat(search): add COLLECT reducer builder + response decode for FT.AGGREGATE

Add FTAggregateCollect options, NewCollectReducer constructor, and
AggregateBuilder.Collect that render the REDUCE COLLECT clause (FIELDS/
DISTINCT/SORTBY/LIMIT) with automatic @-prefixing and arg-count. Add
AggregateRow.Collect to decode the sparse RESP2/RESP3 entry array into a
uniform CollectColumn. Integration-tested against Redis 8.8. Additive:
no existing type or serializer changed.

Requires Redis 8.8+ with search-enable-unstable-features.

* fix(search): validate COLLECT args, harden integration test

- reject empty field names and ASC+DESC combinations client-side,
  matching the FT.AGGREGATE pipeline SORTBY error wording
- gate the integration test on the live server version via INFO;
  the float REDIS_VERSION check parses 8.10 as 8.1 and rots
- restore search-enable-unstable-features, drop the index, and Del
  only the test's own keys instead of FlushDB; remove broken itoa

* docs(example): add FT.AGGREGATE COLLECT reducer example

Runnable example for the COLLECT reducer, mirroring the
search-aggregate-steps layout:

- top-2 products per category via NewCollectReducer (explicit FIELDS
  with SORTBY rating DESC and LIMIT), next to a plain COUNT reducer
- whole documents per brand via AggregateBuilder.Collect with LOAD *
  and FIELDS *
- replies decoded with AggregateRow.Collect (RESP2/RESP3 agnostic)

Verified live against redislabs/client-libs-test:8.8.0. Notes in code
and README: on current unstable builds, SORTBY combined with FIELDS *
drops the sort field from the entries and does not sort - use an
explicit FIELDS list when sorting.

* chore(git): remove binary

---------

Co-authored-by: ofekshenawa <104765379+ofekshenawa@users.noreply.github.com>

example/search-aggregate-collect/README.md

@@ -0,0 +1,41 @@
+# FT.AGGREGATE COLLECT reducer example
+
+This example demonstrates the `COLLECT` reducer for `FT.AGGREGATE`: a
+`GROUPBY` reducer that, within each group, projects a chosen set of fields
+from every row — optionally deduplicated, sorted and limited — and returns
+them as an array of per-entry maps under the reducer alias. A single
+aggregation can therefore return each group **and** its top-N members.
+
+## Prerequisites
+
+Redis 8.8+ with the Search module, listening on `localhost:6379`. COLLECT is
+gated behind unstable search features; the example enables the gate itself
+with `CONFIG SET search-enable-unstable-features yes`.
+
+The client is configured with `Protocol: 2` because the RESP3 response shape
+of `FT.AGGREGATE` is still marked unstable; set `UnstableResp3: true` instead
+if you want to use RESP3. `AggregateRow.Collect` decodes both the RESP2 and
+RESP3 entry shapes, so no other code changes are needed.
+
+## Run
+
+```shell
+go run .
+```
+
+The example:
+
+1. Creates an index `idx:products` over hashes prefixed with `product:`.
+2. Seeds 10 products with `name`, `category`, `brand`, `price`, `rating`.
+3. Runs two aggregations:
+   - **Top-2 per category (options struct)** — `GROUPBY @category` with
+     `REDUCE COLLECT ... FIELDS 2 @name @price SORTBY 2 @rating DESC LIMIT 0 2`,
+     built with `redis.NewCollectReducer` next to a plain `COUNT` reducer.
+   - **Whole documents per brand (builder)** — `LOAD *` → `GROUPBY @brand` →
+     `COLLECT FIELDS *`, expressed with the fluent `AggregateBuilder.Collect`.
+     (On current unstable server builds, `SORTBY` combined with `FIELDS *`
+     drops the sort field from the entries — sort with an explicit field list
+     instead, as in the first query.)
+
+Replies are decoded with `AggregateRow.Collect`, which returns the collected
+entries as `[]map[string]interface{}` regardless of protocol.

example/search-aggregate-collect/go.mod

@@ -0,0 +1,12 @@
+module github.com/redis/go-redis/example/search-aggregate-collect
+
+go 1.24
+
+replace github.com/redis/go-redis/v9 => ../..
+
+require github.com/redis/go-redis/v9 v9.21.0
+
+require (
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	go.uber.org/atomic v1.11.0 // indirect
+)

example/search-aggregate-collect/go.sum

@@ -0,0 +1,20 @@
+github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
+github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
+github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
+github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE=
+github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs=
+github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s=
+go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
+go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
+golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
+golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=

example/search-aggregate-collect/main.go

@@ -0,0 +1,163 @@
+// Example demonstrating the FT.AGGREGATE COLLECT reducer.
+//
+// COLLECT is a GROUPBY reducer that, within each group, projects a chosen
+// set of fields from every row — optionally deduplicated, sorted and
+// limited — and emits them as an array of per-entry maps under the reducer
+// alias. A single aggregation can therefore return each group AND its
+// top-N members.
+//
+// COLLECT requires Redis 8.8+ with unstable search features enabled
+// (CONFIG SET search-enable-unstable-features yes); the example enables
+// the setting itself.
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+
+	"github.com/redis/go-redis/v9"
+)
+
+const indexName = "idx:products"
+
+func main() {
+	ctx := context.Background()
+
+	rdb := redis.NewClient(&redis.Options{
+		Addr:     "localhost:6379",
+		Protocol: 2,
+	})
+	defer rdb.Close()
+
+	// COLLECT is gated behind unstable search features (Redis 8.8+).
+	if err := rdb.ConfigSet(ctx, "search-enable-unstable-features", "yes").Err(); err != nil {
+		log.Fatalf("cannot enable search unstable features (COLLECT needs Redis 8.8+ with search): %v", err)
+	}
+
+	if err := rdb.FlushDB(ctx).Err(); err != nil {
+		log.Fatalf("flushdb: %v", err)
+	}
+	createIndex(ctx, rdb)
+	seedProducts(ctx, rdb)
+
+	fmt.Println("# 1. Top-2 products per category (options struct)")
+	topPerCategory(ctx, rdb)
+
+	fmt.Println("\n# 2. Whole documents per brand (builder, LOAD * + FIELDS *)")
+	documentsPerBrand(ctx, rdb)
+}
+
+func createIndex(ctx context.Context, rdb *redis.Client) {
+	_, err := rdb.FTCreate(ctx, indexName,
+		&redis.FTCreateOptions{OnHash: true, Prefix: []interface{}{"product:"}},
+		&redis.FieldSchema{FieldName: "name", FieldType: redis.SearchFieldTypeText},
+		&redis.FieldSchema{FieldName: "category", FieldType: redis.SearchFieldTypeTag},
+		&redis.FieldSchema{FieldName: "brand", FieldType: redis.SearchFieldTypeTag},
+		&redis.FieldSchema{FieldName: "price", FieldType: redis.SearchFieldTypeNumeric, Sortable: true},
+		&redis.FieldSchema{FieldName: "rating", FieldType: redis.SearchFieldTypeNumeric, Sortable: true},
+	).Result()
+	if err != nil {
+		log.Fatalf("ftcreate: %v", err)
+	}
+}
+
+func seedProducts(ctx context.Context, rdb *redis.Client) {
+	products := []map[string]interface{}{
+		{"name": "Roadster", "category": "bike", "brand": "Velorim", "price": 270, "rating": 4.5},
+		{"name": "Commuter", "category": "bike", "brand": "Velorim", "price": 810, "rating": 4.1},
+		{"name": "Summit Pro", "category": "bike", "brand": "Bicyk", "price": 2300, "rating": 4.9},
+		{"name": "Gravel GT", "category": "bike", "brand": "Bicyk", "price": 430, "rating": 3.8},
+		{"name": "AirFlow", "category": "helmet", "brand": "Nord", "price": 120, "rating": 4.7},
+		{"name": "SafeRide", "category": "helmet", "brand": "Nord", "price": 80, "rating": 4.0},
+		{"name": "ProShell", "category": "helmet", "brand": "Peaknetic", "price": 200, "rating": 4.8},
+		{"name": "BeamMax", "category": "light", "brand": "Peaknetic", "price": 45, "rating": 4.3},
+		{"name": "NightGlow", "category": "light", "brand": "Nord", "price": 30, "rating": 3.9},
+		{"name": "LumenX", "category": "light", "brand": "Velorim", "price": 65, "rating": 4.6},
+	}
+	for i, p := range products {
+		key := fmt.Sprintf("product:%d", i)
+		if err := rdb.HSet(ctx, key, p).Err(); err != nil {
+			log.Fatalf("hset %s: %v", key, err)
+		}
+	}
+}
+
+// topPerCategory groups products by category and, inside each group,
+// collects the name and price of the two highest-rated products:
+//
+//	GROUPBY 1 @category
+//	  REDUCE COUNT 0 AS n
+//	  REDUCE COLLECT 12 FIELDS 2 @name @price SORTBY 2 @rating DESC LIMIT 0 2
+//	  AS top_products
+//
+// NewCollectReducer assembles the token list and computes the argument
+// count; field and sort names may be given with or without the "@" prefix.
+func topPerCategory(ctx context.Context, rdb *redis.Client) {
+	reducer, err := redis.NewCollectReducer(redis.FTAggregateCollect{
+		Fields: []string{"name", "price"},
+		SortBy: []redis.FTAggregateSortBy{{FieldName: "rating", Desc: true}},
+		Limit:  &redis.FTAggregateCollectLimit{Offset: 0, Count: 2},
+		As:     "top_products",
+	})
+	if err != nil {
+		log.Fatalf("collect reducer: %v", err)
+	}
+
+	res, err := rdb.FTAggregateWithArgs(ctx, indexName, "*", &redis.FTAggregateOptions{
+		GroupBy: []redis.FTAggregateGroupBy{{
+			Fields: []interface{}{"@category"},
+			Reduce: []redis.FTAggregateReducer{
+				{Reducer: redis.SearchCount, As: "n"},
+				reducer,
+			},
+		}},
+	}).Result()
+	if err != nil {
+		log.Fatalf("aggregate: %v", err)
+	}
+
+	for _, row := range res.Rows {
+		fmt.Printf("  category=%v n=%v\n", row.Fields["category"], row.Fields["n"])
+		printEntries(row, "top_products")
+	}
+}
+
+// documentsPerBrand expresses the same reducer through the fluent builder.
+// FIELDS * projects every field present in the pipeline at the COLLECT
+// stage, so it is paired with an upstream LOAD * to collect complete
+// documents. Entry order is unspecified without a SORTBY; on current
+// unstable server builds, SORTBY combined with FIELDS * drops the sort
+// field from the entries, so sort with an explicit Fields list instead
+// (as in topPerCategory).
+func documentsPerBrand(ctx context.Context, rdb *redis.Client) {
+	res, err := rdb.NewAggregateBuilder(ctx, indexName, "*").
+		LoadAll().
+		GroupBy("@brand").
+		Collect(redis.FTAggregateCollect{
+			FieldsAll: true,
+			As:        "products",
+		}).
+		Run()
+	if err != nil {
+		log.Fatalf("aggregate: %v", err)
+	}
+
+	for _, row := range res.Rows {
+		fmt.Printf("  brand=%v\n", row.Fields["brand"])
+		printEntries(row, "products")
+	}
+}
+
+// printEntries decodes the COLLECT column stored under alias with
+// AggregateRow.Collect, which hides the RESP2/RESP3 representation
+// difference, and prints one line per collected entry.
+func printEntries(row redis.AggregateRow, alias string) {
+	entries, err := row.Collect(alias)
+	if err != nil {
+		log.Fatalf("decode %s: %v", alias, err)
+	}
+	for _, e := range entries {
+		fmt.Printf("    %v\n", e)
+	}
+}

search_builders.go

@@ -313,6 +313,31 @@ func (b *AggregateBuilder) ReduceAs(fn SearchAggregator, alias string, args ...i
 	return b
 }
 
+// Collect adds a REDUCE COLLECT clause to the last step, which must be a
+// GROUPBY. The COLLECT options (FIELDS/DISTINCT/SORTBY/LIMIT/AS) are rendered
+// and the argument count is computed automatically; field and sort names are
+// normalized to a single "@" prefix. Set FTAggregateCollect.As to alias the
+// output column.
+//
+// If the last step is not a GROUPBY, or the options are invalid (no FIELDS
+// selector), Run returns the recorded error without issuing the command.
+// COLLECT requires Redis 8.8+ with unstable features enabled.
+func (b *AggregateBuilder) Collect(o FTAggregateCollect) *AggregateBuilder {
+	n := len(b.options.Steps)
+	if n == 0 || b.options.Steps[n-1].GroupBy == nil {
+		b.setErr(fmt.Errorf("FT.AGGREGATE: Collect must follow a GroupBy step"))
+		return b
+	}
+	reducer, err := NewCollectReducer(o)
+	if err != nil {
+		b.setErr(err)
+		return b
+	}
+	g := b.options.Steps[n-1].GroupBy
+	g.Reduce = append(g.Reduce, reducer)
+	return b
+}
+
 // SortBy adds SORTBY <field> ASC|DESC. Consecutive SortBy calls (with no
 // other step in between) are merged into a single SORTBY clause so fields
 // act as tiebreakers. A SortBy call after a non-SortBy step starts a new

search_collect.go

@@ -0,0 +1,235 @@
+package redis
+
+import (
+	"fmt"
+	"strings"
+)
+
+// ----------------------
+// FT.AGGREGATE COLLECT reducer
+// ----------------------
+//
+// COLLECT is a GROUPBY reducer for FT.AGGREGATE (Redis 8.8+, gated behind
+// search-enable-unstable-features). Within each group it projects a chosen
+// set of fields from every row, optionally deduplicates, sorts, and limits
+// them, and emits the result as an array of per-entry maps under the reducer
+// alias.
+//
+// COLLECT is not a standalone command; it is a REDUCE clause inside
+// FT.AGGREGATE. The helpers below assemble the reducer token list and compute
+// its argument count, so callers do not have to hand-write FIELDS/SORTBY/LIMIT
+// tokens or remember to @-prefix every name.
+
+// FTAggregateCollect describes a COLLECT reducer. It is rendered into a
+// standard FTAggregateReducer via NewCollectReducer, or appended to a builder
+// via AggregateBuilder.Collect.
+//
+// Field and sort names may be supplied with or without a leading "@"; each is
+// normalized to a single "@<name>" on the wire. Output map keys returned by
+// the server are the bare names (see AggregateRow.Collect).
+type FTAggregateCollect struct {
+	// FieldsAll emits FIELDS *, projecting every field present in the
+	// pipeline at the COLLECT stage. It is not a whole-document fetch; pair
+	// it with an upstream LOAD * to collect complete documents. FieldsAll
+	// takes precedence over Fields when both are set.
+	FieldsAll bool
+
+	// Fields is the explicit list of fields to project (FIELDS <n> @f ...).
+	// Ignored when FieldsAll is true. Exactly one of FieldsAll or a non-empty
+	// Fields must be set.
+	Fields []string
+
+	// Distinct emits DISTINCT, deduplicating entries with identical projected
+	// fields.
+	//
+	// NOTE: DISTINCT is specified by the product but not yet implemented by
+	// the server. Sending it currently produces a server error. The option is
+	// kept for forward compatibility; leave it false unless the target server
+	// supports it.
+	Distinct bool
+
+	// SortBy orders entries within each group. Direction defaults to ASC when
+	// neither Asc nor Desc is set. With Limit, SORTBY acts as a top-N
+	// selection. Reuses FTAggregateSortBy for consistency with the rest of the
+	// aggregate API.
+	SortBy []FTAggregateSortBy
+
+	// Limit returns at most Count entries per group after skipping Offset.
+	// nil means no LIMIT clause (distinct from LIMIT 0 0).
+	Limit *FTAggregateCollectLimit
+
+	// As sets the reducer output column name (AS <alias>). It is emitted
+	// outside the reducer argument count.
+	As string
+}
+
+// FTAggregateCollectLimit is the LIMIT <offset> <count> clause of a COLLECT
+// reducer. Numeric bounds are enforced by the server, not the client.
+type FTAggregateCollectLimit struct {
+	Offset int
+	Count  int
+}
+
+// ensureAtPrefix normalizes a field or sort name to exactly one leading "@",
+// collapsing any number of leading "@" (including none) to a single prefix.
+func ensureAtPrefix(name string) string {
+	return "@" + strings.TrimLeft(name, "@")
+}
+
+// buildCollectArgs renders a FTAggregateCollect into the reducer argument
+// token list (everything after "REDUCE COLLECT <narg>", excluding AS <alias>).
+// The serializer computes <narg> as len(args), which matches the COLLECT
+// contract: narg counts every FIELDS/DISTINCT/SORTBY/LIMIT token.
+func buildCollectArgs(o FTAggregateCollect) ([]interface{}, error) {
+	args := make([]interface{}, 0, 8)
+
+	// FIELDS (required): either * or a counted list of @-names.
+	switch {
+	case o.FieldsAll:
+		args = append(args, "FIELDS", "*")
+	case len(o.Fields) > 0:
+		args = append(args, "FIELDS", len(o.Fields))
+		for _, f := range o.Fields {
+			if strings.TrimLeft(f, "@") == "" {
+				return nil, fmt.Errorf("redis: FT.AGGREGATE COLLECT: empty field name in Fields")
+			}
+			args = append(args, ensureAtPrefix(f))
+		}
+	default:
+		return nil, fmt.Errorf("redis: FT.AGGREGATE COLLECT requires FieldsAll or a non-empty Fields list")
+	}
+
+	// DISTINCT (optional, forward-compatible).
+	if o.Distinct {
+		args = append(args, "DISTINCT")
+	}
+
+	// SORTBY (optional). sort_narg counts each field plus its optional
+	// direction token.
+	if len(o.SortBy) > 0 {
+		sortTokens := make([]interface{}, 0, len(o.SortBy)*2)
+		for _, s := range o.SortBy {
+			if strings.TrimLeft(s.FieldName, "@") == "" {
+				return nil, fmt.Errorf("redis: FT.AGGREGATE COLLECT: empty field name in SortBy")
+			}
+			if s.Asc && s.Desc {
+				return nil, fmt.Errorf("redis: FT.AGGREGATE COLLECT: ASC and DESC are mutually exclusive")
+			}
+			sortTokens = append(sortTokens, ensureAtPrefix(s.FieldName))
+			switch {
+			case s.Desc:
+				sortTokens = append(sortTokens, "DESC")
+			case s.Asc:
+				sortTokens = append(sortTokens, "ASC")
+				// neither set: ASC is the server default; emit nothing.
+			}
+		}
+		args = append(args, "SORTBY", len(sortTokens))
+		args = append(args, sortTokens...)
+	}
+
+	// LIMIT (optional).
+	if o.Limit != nil {
+		args = append(args, "LIMIT", o.Limit.Offset, o.Limit.Count)
+	}
+
+	return args, nil
+}
+
+// NewCollectReducer builds a COLLECT FTAggregateReducer for use with
+// FTAggregateOptions.GroupBy[i].Reduce. It normalizes field/sort names and
+// computes the argument count automatically.
+//
+// It returns an error only for local API misuse: a missing FIELDS selector, an
+// empty field name (in Fields or SortBy), or a SortBy entry with both Asc and
+// Desc set. Numeric bounds and the unstable-features gate are enforced by the
+// server and surface unchanged through the command reply.
+func NewCollectReducer(o FTAggregateCollect) (FTAggregateReducer, error) {
+	args, err := buildCollectArgs(o)
+	if err != nil {
+		return FTAggregateReducer{}, err
+	}
+	return FTAggregateReducer{Reducer: SearchCollect, Args: args, As: o.As}, nil
+}
+
+// ----------------------
+// COLLECT response decoding
+// ----------------------
+
+// CollectEntry is a single collected row: a sparse map of bare field name to
+// value. A field absent from a row is omitted from its entry (no NULL
+// placeholder), so entries in the same column may have different key sets.
+type CollectEntry = map[string]interface{}
+
+// CollectColumn is the value stored under a COLLECT reducer alias: the ordered
+// list of collected entries for a group.
+type CollectColumn = []CollectEntry
+
+// Collect decodes the COLLECT reducer column stored under alias in this row
+// into a uniform CollectColumn, hiding the RESP2/RESP3 representation
+// difference (RESP3 entries are maps; RESP2 entries are flat key/value
+// arrays).
+//
+// It returns (nil, nil) when the alias is absent from the row. Entry order is
+// preserved as returned by the server; it is meaningful only when the COLLECT
+// reducer was given a SORTBY.
+func (r AggregateRow) Collect(alias string) (CollectColumn, error) {
+	v, ok := r.Fields[alias]
+	if !ok {
+		return nil, nil
+	}
+	return parseCollectValue(v)
+}
+
+// parseCollectValue decodes a raw COLLECT alias value (an array of entries)
+// into a CollectColumn.
+func parseCollectValue(v interface{}) (CollectColumn, error) {
+	if v == nil {
+		return nil, nil
+	}
+	arr, ok := v.([]interface{})
+	if !ok {
+		return nil, fmt.Errorf("redis: COLLECT value has type %T, want array of entries", v)
+	}
+	out := make(CollectColumn, 0, len(arr))
+	for i, e := range arr {
+		entry, err := parseCollectEntry(e)
+		if err != nil {
+			return nil, fmt.Errorf("redis: COLLECT entry %d: %w", i, err)
+		}
+		out = append(out, entry)
+	}
+	return out, nil
+}
+
+// parseCollectEntry decodes a single collected entry from either the RESP3
+// map form or the RESP2 flat key/value array form into a CollectEntry. Keys
+// are passed through as-is: the server already returns them without the "@"
+// prefix.
+func parseCollectEntry(e interface{}) (CollectEntry, error) {
+	switch m := e.(type) {
+	case map[interface{}]interface{}: // RESP3
+		out := make(CollectEntry, len(m))
+		for k, val := range m {
+			out[fmt.Sprint(k)] = val
+		}
+		return out, nil
+	case map[string]interface{}: // already string-keyed
+		return m, nil
+	case []interface{}: // RESP2 flat [field, value, field, value, ...]
+		if len(m)%2 != 0 {
+			return nil, fmt.Errorf("odd-length key/value array of length %d", len(m))
+		}
+		out := make(CollectEntry, len(m)/2)
+		for i := 0; i < len(m); i += 2 {
+			key, ok := m[i].(string)
+			if !ok {
+				key = fmt.Sprint(m[i])
+			}
+			out[key] = m[i+1]
+		}
+		return out, nil
+	default:
+		return nil, fmt.Errorf("unexpected type %T, want map or key/value array", e)
+	}
+}

search_collect_integration_test.go

@@ -0,0 +1,256 @@
+package redis_test
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"sort"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/redis/go-redis/v9"
+)
+
+// collectTestAddr is the standalone Redis used by the COLLECT integration
+// test. Defaults to :6379 (the repo's standard test server) and can be
+// overridden, e.g. when a Redis 8.8 + search-enable-unstable-features instance
+// runs on a different port:
+//
+//	REDIS_COLLECT_TEST_ADDR=localhost:6399 go test -run TestFTAggregateCollect_Integration
+func collectTestAddr() string {
+	if a := os.Getenv("REDIS_COLLECT_TEST_ADDR"); a != "" {
+		return a
+	}
+	return ":6379"
+}
+
+// TestFTAggregateCollect_Integration exercises the COLLECT reducer end-to-end
+// against a live server, through both the options-struct (NewCollectReducer)
+// and fluent-builder (AggregateBuilder.Collect) surfaces, and decodes the
+// reply with AggregateRow.Collect under both RESP2 and RESP3.
+//
+// COLLECT requires Redis 8.8+ with search-enable-unstable-features enabled.
+// The test skips if the server is unreachable, lacks the search module, or
+// does not allow enabling unstable features.
+func TestFTAggregateCollect_Integration(t *testing.T) {
+	// COLLECT reducer ships in Redis 8.8. Older search builds accept the
+	// search-enable-unstable-features config but still reject the reducer
+	// ("No such reducer: COLLECT"), so gate on the live server's version
+	// (INFO server redis_version) rather than on the config toggle or the
+	// REDIS_VERSION env variable (whose float form cannot tell 8.10 from 8.1
+	// and defaults when unset).
+	probeCtx := context.Background()
+	probe := redis.NewClient(&redis.Options{Addr: collectTestAddr()})
+	t.Cleanup(func() { _ = probe.Close() })
+	if err := probe.Ping(probeCtx).Err(); err != nil {
+		t.Skipf("redis not reachable at %s: %v", collectTestAddr(), err)
+	}
+	major, minor, err := collectServerVersion(probeCtx, probe)
+	if err != nil {
+		t.Fatalf("cannot determine server version: %v", err)
+	}
+	if major < 8 || (major == 8 && minor < 8) {
+		t.Skipf("COLLECT requires Redis 8.8+ (server is %d.%d)", major, minor)
+	}
+	for _, proto := range []int{2, 3} {
+		proto := proto
+		t.Run(fmt.Sprintf("RESP%d", proto), func(t *testing.T) {
+			ctx := context.Background()
+			opt := &redis.Options{Addr: collectTestAddr(), Protocol: proto}
+			if proto == 3 {
+				opt.UnstableResp3 = true
+			}
+			client := redis.NewClient(opt)
+			t.Cleanup(func() { _ = client.Close() })
+
+			if err := client.Ping(ctx).Err(); err != nil {
+				t.Skipf("redis not reachable at %s: %v", collectTestAddr(), err)
+			}
+			// COLLECT is gated behind unstable features; skip if we cannot
+			// enable it. Remember the prior value and restore it on cleanup so
+			// the test does not leak config changes into the shared server.
+			const unstableCfg = "search-enable-unstable-features"
+			prevCfg, err := client.ConfigGet(ctx, unstableCfg).Result()
+			if err != nil {
+				t.Skipf("cannot read %s (need Redis 8.8+ with search): %v", unstableCfg, err)
+			}
+			if err := client.ConfigSet(ctx, unstableCfg, "yes").Err(); err != nil {
+				t.Skipf("cannot enable search unstable features (need Redis 8.8+ with search): %v", err)
+			}
+			t.Cleanup(func() {
+				if prev, ok := prevCfg[unstableCfg]; ok {
+					_ = client.ConfigSet(ctx, unstableCfg, prev).Err()
+				}
+			})
+
+			// Clear only this test's keys (never FlushDB: the server is shared).
+			docKeys := []string{"fruit:1", "fruit:2", "fruit:3", "fruit:4"}
+			if err := client.Del(ctx, docKeys...).Err(); err != nil {
+				t.Fatalf("Del %v: %v", docKeys, err)
+			}
+
+			const idx = "idx_collect_it"
+			if err := client.Do(ctx,
+				"FT.CREATE", idx, "ON", "HASH", "PREFIX", 1, "fruit:",
+				"SCHEMA", "color", "TAG", "name", "TEXT", "sweetness", "NUMERIC", "SORTABLE",
+			).Err(); err != nil {
+				if strings.Contains(strings.ToUpper(err.Error()), "UNKNOWN COMMAND") {
+					t.Skipf("search module not loaded: %v", err)
+				}
+				t.Fatalf("FT.CREATE: %v", err)
+			}
+			// Drop the index and its documents (DD) on cleanup.
+			t.Cleanup(func() {
+				_ = client.FTDropIndexWithArgs(ctx, idx,
+					&redis.FTDropIndexOptions{DeleteDocs: true}).Err()
+			})
+
+			docs := []struct {
+				key, color, name string
+				sweetness        interface{}
+			}{
+				{"fruit:1", "red", "apple", 4},
+				{"fruit:2", "red", "strawberry", 3},
+				{"fruit:3", "yellow", "banana", 4},
+				{"fruit:4", "yellow", "lemon", nil}, // sparse: no sweetness
+			}
+			for _, d := range docs {
+				vals := []interface{}{"color", d.color, "name", d.name}
+				if d.sweetness != nil {
+					vals = append(vals, "sweetness", d.sweetness)
+				}
+				if err := client.HSet(ctx, d.key, vals...).Err(); err != nil {
+					t.Fatalf("HSet %s: %v", d.key, err)
+				}
+			}
+
+			// --- Surface 1: options-struct via NewCollectReducer ---
+			reducer, err := redis.NewCollectReducer(redis.FTAggregateCollect{
+				Fields: []string{"name", "sweetness"}, // no @ prefix: must be normalized
+				SortBy: []redis.FTAggregateSortBy{{FieldName: "sweetness", Desc: true}},
+				As:     "fruits",
+			})
+			if err != nil {
+				t.Fatalf("NewCollectReducer: %v", err)
+			}
+			res, err := client.FTAggregateWithArgs(ctx, idx, "*", &redis.FTAggregateOptions{
+				GroupBy: []redis.FTAggregateGroupBy{{
+					Fields: []interface{}{"@color"},
+					Reduce: []redis.FTAggregateReducer{reducer},
+				}},
+			}).Result()
+			if err != nil {
+				t.Fatalf("FTAggregateWithArgs COLLECT: %v", err)
+			}
+			assertCollectGroups(t, res)
+
+			// --- Surface 2: fluent builder ---
+			res2, err := client.NewAggregateBuilder(ctx, idx, "*").
+				GroupBy("@color").
+				Collect(redis.FTAggregateCollect{
+					FieldsAll: true,
+					SortBy:    []redis.FTAggregateSortBy{{FieldName: "sweetness", Desc: true}},
+					As:        "fruits",
+				}).
+				Run()
+			if err != nil {
+				t.Fatalf("builder COLLECT: %v", err)
+			}
+			// FIELDS * projects pipeline fields (color group key + name + sweetness).
+			if len(res2.Rows) != 2 {
+				t.Fatalf("builder: got %d groups, want 2", len(res2.Rows))
+			}
+			for _, row := range res2.Rows {
+				col, err := row.Collect("fruits")
+				if err != nil {
+					t.Fatalf("builder row.Collect: %v", err)
+				}
+				if len(col) == 0 {
+					t.Fatalf("builder: empty collect column for row %v", row.Fields)
+				}
+			}
+		})
+	}
+}
+
+// assertCollectGroups verifies the two color groups, ordering within a group,
+// and the sparse entry (lemon has no sweetness).
+func assertCollectGroups(t *testing.T, res *redis.FTAggregateResult) {
+	t.Helper()
+	if len(res.Rows) != 2 {
+		t.Fatalf("got %d groups, want 2 (red, yellow)", len(res.Rows))
+	}
+
+	byColor := map[string]redis.CollectColumn{}
+	for _, row := range res.Rows {
+		color, _ := row.Fields["color"].(string)
+		col, err := row.Collect("fruits")
+		if err != nil {
+			t.Fatalf("row.Collect(fruits) for color=%q: %v", color, err)
+		}
+		byColor[color] = col
+	}
+
+	red := byColor["red"]
+	if len(red) != 2 {
+		t.Fatalf("red group: got %d entries, want 2", len(red))
+	}
+	// SORTBY sweetness DESC: apple(4) before strawberry(3).
+	if got := red[0]["name"]; got != "apple" {
+		t.Fatalf("red[0].name = %v, want apple (DESC sweetness order)", got)
+	}
+	if got := red[1]["name"]; got != "strawberry" {
+		t.Fatalf("red[1].name = %v, want strawberry", got)
+	}
+
+	yellow := byColor["yellow"]
+	if len(yellow) != 2 {
+		t.Fatalf("yellow group: got %d entries, want 2", len(yellow))
+	}
+	// Sparse: the lemon entry must omit sweetness (no NULL placeholder).
+	names := []string{yellow[0]["name"].(string), yellow[1]["name"].(string)}
+	sort.Strings(names)
+	if names[0] != "banana" || names[1] != "lemon" {
+		t.Fatalf("yellow names = %v, want [banana lemon]", names)
+	}
+	for _, e := range yellow {
+		if e["name"] == "lemon" {
+			if _, ok := e["sweetness"]; ok {
+				t.Fatalf("lemon entry must be sparse (no sweetness), got %v", e)
+			}
+		}
+	}
+}
+
+// collectServerVersion returns the major and minor components of the live
+// server's redis_version, taken from INFO server. Unlike the float-typed
+// RedisVersion (parsed from the REDIS_VERSION env variable), integer
+// major/minor comparison keeps 8.10 > 8.8 and reflects the server actually
+// answering, not the environment.
+func collectServerVersion(ctx context.Context, client *redis.Client) (major, minor int, err error) {
+	info, err := client.Info(ctx, "server").Result()
+	if err != nil {
+		return 0, 0, err
+	}
+	for _, line := range strings.Split(info, "\n") {
+		v, ok := strings.CutPrefix(strings.TrimSpace(line), "redis_version:")
+		if !ok {
+			continue
+		}
+		parts := strings.SplitN(v, ".", 3)
+		if len(parts) < 2 {
+			return 0, 0, fmt.Errorf("unexpected redis_version %q in INFO server", v)
+		}
+		major, err = strconv.Atoi(parts[0])
+		if err != nil {
+			return 0, 0, fmt.Errorf("unexpected redis_version %q in INFO server: %w", v, err)
+		}
+		minor, err = strconv.Atoi(parts[1])
+		if err != nil {
+			return 0, 0, fmt.Errorf("unexpected redis_version %q in INFO server: %w", v, err)
+		}
+		return major, minor, nil
+	}
+	return 0, 0, fmt.Errorf("redis_version not found in INFO server reply")
+}

search_collect_test.go

@@ -0,0 +1,297 @@
+package redis
+
+import (
+	"context"
+	"reflect"
+	"testing"
+)
+
+func TestSearchCollect_String(t *testing.T) {
+	if got := SearchCollect.String(); got != "COLLECT" {
+		t.Fatalf("SearchCollect.String() = %q, want %q", got, "COLLECT")
+	}
+}
+
+// TestNewCollectReducer_Wire is the wire oracle: an options-struct COLLECT
+// query must serialize to the exact REDUCE COLLECT clause, with the argument
+// count computed from the assembled tokens and every name @-prefixed. Field
+// names are supplied with mixed prefixes to exercise normalization.
+func TestNewCollectReducer_Wire(t *testing.T) {
+	m := &mockCmdable{}
+	c := m.asCmdable()
+
+	reducer, err := NewCollectReducer(FTAggregateCollect{
+		Fields: []string{"title", "@rating", "year"}, // mixed @-prefix
+		SortBy: []FTAggregateSortBy{{FieldName: "rating", Desc: true}},
+		Limit:  &FTAggregateCollectLimit{Offset: 0, Count: 5},
+		As:     "top_movies",
+	})
+	if err != nil {
+		t.Fatalf("NewCollectReducer: %v", err)
+	}
+
+	opts := &FTAggregateOptions{
+		GroupBy: []FTAggregateGroupBy{{
+			Fields: []interface{}{"@genre"},
+			Reduce: []FTAggregateReducer{reducer},
+		}},
+	}
+	cmd := c.FTAggregateWithArgs(context.Background(), "idx", "*", opts)
+	if cmd.Err() != nil {
+		t.Fatalf("unexpected error: %v", cmd.Err())
+	}
+
+	want := []interface{}{
+		"FT.AGGREGATE", "idx", "*",
+		"GROUPBY", 1, "@genre",
+		"REDUCE", "COLLECT", 12,
+		"FIELDS", 3, "@title", "@rating", "@year",
+		"SORTBY", 2, "@rating", "DESC",
+		"LIMIT", 0, 5,
+		"AS", "top_movies",
+		"DIALECT", 2,
+	}
+	if !reflect.DeepEqual(cmd.args, want) {
+		t.Fatalf("args mismatch:\n got:  %v\n want: %v", cmd.args, want)
+	}
+	// narg must equal the token count.
+	if got := len(reducer.Args); got != 12 {
+		t.Fatalf("narg = %d, want 12", got)
+	}
+}
+
+func TestBuildCollectArgs(t *testing.T) {
+	tests := []struct {
+		name    string
+		in      FTAggregateCollect
+		want    []interface{}
+		wantErr bool
+	}{
+		{
+			name: "fields all",
+			in:   FTAggregateCollect{FieldsAll: true},
+			want: []interface{}{"FIELDS", "*"},
+		},
+		{
+			name: "fields all wins over explicit list",
+			in:   FTAggregateCollect{FieldsAll: true, Fields: []string{"x"}},
+			want: []interface{}{"FIELDS", "*"},
+		},
+		{
+			name: "explicit fields normalized",
+			in:   FTAggregateCollect{Fields: []string{"a", "@b", "@@c"}},
+			want: []interface{}{"FIELDS", 3, "@a", "@b", "@c"},
+		},
+		{
+			name: "sortby default asc emits nothing",
+			in: FTAggregateCollect{
+				Fields: []string{"a"},
+				SortBy: []FTAggregateSortBy{{FieldName: "a"}},
+			},
+			want: []interface{}{"FIELDS", 1, "@a", "SORTBY", 1, "@a"},
+		},
+		{
+			name: "sortby explicit asc emits ASC",
+			in: FTAggregateCollect{
+				Fields: []string{"a"},
+				SortBy: []FTAggregateSortBy{{FieldName: "a", Asc: true}},
+			},
+			want: []interface{}{"FIELDS", 1, "@a", "SORTBY", 2, "@a", "ASC"},
+		},
+		{
+			name: "sortby multi keys mixed direction",
+			in: FTAggregateCollect{
+				Fields: []string{"a"},
+				SortBy: []FTAggregateSortBy{{FieldName: "a", Desc: true}, {FieldName: "b"}},
+			},
+			want: []interface{}{"FIELDS", 1, "@a", "SORTBY", 3, "@a", "DESC", "@b"},
+		},
+		{
+			name: "distinct emitted",
+			in:   FTAggregateCollect{FieldsAll: true, Distinct: true},
+			want: []interface{}{"FIELDS", "*", "DISTINCT"},
+		},
+		{
+			name: "limit emitted",
+			in:   FTAggregateCollect{FieldsAll: true, Limit: &FTAggregateCollectLimit{Offset: 2, Count: 7}},
+			want: []interface{}{"FIELDS", "*", "LIMIT", 2, 7},
+		},
+		{
+			name:    "no fields selector is an error",
+			in:      FTAggregateCollect{},
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := buildCollectArgs(tt.in)
+			if tt.wantErr {
+				if err == nil {
+					t.Fatalf("expected error, got args %v", got)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Fatalf("args mismatch:\n got:  %v\n want: %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestEnsureAtPrefix(t *testing.T) {
+	cases := map[string]string{
+		"name":    "@name",
+		"@name":   "@name",
+		"@@name":  "@name",
+		"@@@name": "@name",
+		"__key":   "@__key",
+		"@__key":  "@__key",
+		"":        "@",
+	}
+	for in, want := range cases {
+		if got := ensureAtPrefix(in); got != want {
+			t.Errorf("ensureAtPrefix(%q) = %q, want %q", in, got, want)
+		}
+	}
+}
+
+func TestNewCollectReducer_Error(t *testing.T) {
+	if _, err := NewCollectReducer(FTAggregateCollect{}); err == nil {
+		t.Fatal("expected error for missing FIELDS selector")
+	}
+}
+
+// TestAggregateBuilder_Collect checks the builder appends a correctly rendered
+// COLLECT reducer to the last GROUPBY step. The wire serialization itself is
+// covered by TestNewCollectReducer_Wire (same serializer path).
+func TestAggregateBuilder_Collect(t *testing.T) {
+	b := &AggregateBuilder{options: &FTAggregateOptions{LimitOffset: -1}}
+	b.GroupBy("@genre").Collect(FTAggregateCollect{
+		Fields: []string{"title"},
+		Limit:  &FTAggregateCollectLimit{Offset: 0, Count: 3},
+		As:     "top",
+	})
+	if b.err != nil {
+		t.Fatalf("unexpected builder error: %v", b.err)
+	}
+	if n := len(b.options.Steps); n != 1 {
+		t.Fatalf("steps = %d, want 1", n)
+	}
+	g := b.options.Steps[0].GroupBy
+	if g == nil || len(g.Reduce) != 1 {
+		t.Fatalf("expected one reducer on the GroupBy step, got %+v", g)
+	}
+	r := g.Reduce[0]
+	if r.Reducer != SearchCollect || r.As != "top" {
+		t.Fatalf("reducer = %v as=%q, want COLLECT as=top", r.Reducer, r.As)
+	}
+	want := []interface{}{"FIELDS", 1, "@title", "LIMIT", 0, 3}
+	if !reflect.DeepEqual(r.Args, want) {
+		t.Fatalf("reducer args mismatch:\n got:  %v\n want: %v", r.Args, want)
+	}
+}
+
+func TestAggregateBuilder_Collect_Errors(t *testing.T) {
+	t.Run("without groupby", func(t *testing.T) {
+		b := &AggregateBuilder{options: &FTAggregateOptions{LimitOffset: -1}}
+		b.Collect(FTAggregateCollect{FieldsAll: true})
+		if b.err == nil {
+			t.Fatal("expected error: Collect must follow a GroupBy step")
+		}
+	})
+	t.Run("invalid options", func(t *testing.T) {
+		b := &AggregateBuilder{options: &FTAggregateOptions{LimitOffset: -1}}
+		b.GroupBy("@genre").Collect(FTAggregateCollect{})
+		if b.err == nil {
+			t.Fatal("expected error: missing FIELDS selector")
+		}
+	})
+}
+
+// TestAggregateRow_Collect_RESP3 decodes the RESP3 map form, including a sparse
+// entry that omits a projected field.
+func TestAggregateRow_Collect_RESP3(t *testing.T) {
+	row := AggregateRow{Fields: map[string]interface{}{
+		"fruits": []interface{}{
+			map[interface{}]interface{}{"name": "apple", "sweetness": "4"},
+			map[interface{}]interface{}{"name": "strawberry"}, // sparse: no sweetness
+		},
+	}}
+
+	col, err := row.Collect("fruits")
+	if err != nil {
+		t.Fatalf("Collect: %v", err)
+	}
+	if len(col) != 2 {
+		t.Fatalf("len = %d, want 2", len(col))
+	}
+	if col[0]["name"] != "apple" || col[0]["sweetness"] != "4" {
+		t.Fatalf("entry 0 = %v", col[0])
+	}
+	if col[1]["name"] != "strawberry" {
+		t.Fatalf("entry 1 name = %v", col[1]["name"])
+	}
+	if _, ok := col[1]["sweetness"]; ok {
+		t.Fatalf("entry 1 should be sparse, has sweetness: %v", col[1])
+	}
+}
+
+// TestAggregateRow_Collect_RESP2 decodes the RESP2 flat key/value array form.
+func TestAggregateRow_Collect_RESP2(t *testing.T) {
+	row := AggregateRow{Fields: map[string]interface{}{
+		"fruits": []interface{}{
+			[]interface{}{"name", "apple", "sweetness", "4"},
+			[]interface{}{"name", "strawberry"}, // sparse
+		},
+	}}
+
+	col, err := row.Collect("fruits")
+	if err != nil {
+		t.Fatalf("Collect: %v", err)
+	}
+	if len(col) != 2 {
+		t.Fatalf("len = %d, want 2", len(col))
+	}
+	if col[0]["name"] != "apple" || col[0]["sweetness"] != "4" {
+		t.Fatalf("entry 0 = %v", col[0])
+	}
+	if _, ok := col[1]["sweetness"]; ok {
+		t.Fatalf("entry 1 should be sparse: %v", col[1])
+	}
+}
+
+func TestAggregateRow_Collect_EdgeCases(t *testing.T) {
+	t.Run("absent alias", func(t *testing.T) {
+		row := AggregateRow{Fields: map[string]interface{}{}}
+		col, err := row.Collect("missing")
+		if err != nil || col != nil {
+			t.Fatalf("got (%v, %v), want (nil, nil)", col, err)
+		}
+	})
+	t.Run("non-array value", func(t *testing.T) {
+		row := AggregateRow{Fields: map[string]interface{}{"x": "scalar"}}
+		if _, err := row.Collect("x"); err == nil {
+			t.Fatal("expected error for non-array COLLECT value")
+		}
+	})
+	t.Run("odd-length resp2 entry", func(t *testing.T) {
+		row := AggregateRow{Fields: map[string]interface{}{
+			"x": []interface{}{[]interface{}{"name", "apple", "dangling"}},
+		}}
+		if _, err := row.Collect("x"); err == nil {
+			t.Fatal("expected error for odd-length key/value entry")
+		}
+	})
+	t.Run("bad entry type", func(t *testing.T) {
+		row := AggregateRow{Fields: map[string]interface{}{
+			"x": []interface{}{42},
+		}}
+		if _, err := row.Collect("x"); err == nil {
+			t.Fatal("expected error for non-map/non-array entry")
+		}
+	})
+}

search_commands.go

@@ -165,6 +165,12 @@ const (
 	SearchToList
 	SearchFirstValue
 	SearchRandomSample
+	// SearchCollect is the COLLECT reducer for FT.AGGREGATE. Within each
+	// GROUPBY group it projects a chosen set of fields from every row and
+	// emits them as an array of per-entry maps under the reducer alias.
+	// Requires Redis 8.8+ with unstable features enabled
+	// (CONFIG SET search-enable-unstable-features yes).
+	SearchCollect
 )
 
 func (a SearchAggregator) String() string {
@@ -195,6 +201,8 @@ func (a SearchAggregator) String() string {
 		return "FIRST_VALUE"
 	case SearchRandomSample:
 		return "RANDOM_SAMPLE"
+	case SearchCollect:
+		return "COLLECT"
 	default:
 		return ""
 	}