From 8ba2e2a20fc8ec415255a3776db154a333b67a1e Mon Sep 17 00:00:00 2001 From: Nedyalko Dyakov <1547186+ndyakov@users.noreply.github.com> Date: Wed, 29 Jul 2026 13:20:27 +0300 Subject: [PATCH] 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 | 41 +++ example/search-aggregate-collect/go.mod | 12 + example/search-aggregate-collect/go.sum | 20 ++ example/search-aggregate-collect/main.go | 163 +++++++++++ search_builders.go | 25 ++ search_collect.go | 235 ++++++++++++++++ search_collect_integration_test.go | 256 ++++++++++++++++++ search_collect_test.go | 297 +++++++++++++++++++++ search_commands.go | 8 + 9 files changed, 1057 insertions(+) create mode 100644 example/search-aggregate-collect/README.md create mode 100644 example/search-aggregate-collect/go.mod create mode 100644 example/search-aggregate-collect/go.sum create mode 100644 example/search-aggregate-collect/main.go create mode 100644 search_collect.go create mode 100644 search_collect_integration_test.go create mode 100644 search_collect_test.go diff --git a/example/search-aggregate-collect/README.md b/example/search-aggregate-collect/README.md new file mode 100644 index 00000000..a9c3bf3b --- /dev/null +++ b/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. diff --git a/example/search-aggregate-collect/go.mod b/example/search-aggregate-collect/go.mod new file mode 100644 index 00000000..f520c741 --- /dev/null +++ b/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 +) diff --git a/example/search-aggregate-collect/go.sum b/example/search-aggregate-collect/go.sum new file mode 100644 index 00000000..e9c28ca0 --- /dev/null +++ b/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= diff --git a/example/search-aggregate-collect/main.go b/example/search-aggregate-collect/main.go new file mode 100644 index 00000000..d2afed49 --- /dev/null +++ b/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) + } +} diff --git a/search_builders.go b/search_builders.go index a6c6718c..094441c7 100644 --- a/search_builders.go +++ b/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 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 diff --git a/search_collect.go b/search_collect.go new file mode 100644 index 00000000..48389a0c --- /dev/null +++ b/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 "@" 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 @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 ). It is emitted + // outside the reducer argument count. + As string +} + +// FTAggregateCollectLimit is the LIMIT 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 ", excluding AS ). +// The serializer computes 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) + } +} diff --git a/search_collect_integration_test.go b/search_collect_integration_test.go new file mode 100644 index 00000000..4be05535 --- /dev/null +++ b/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") +} diff --git a/search_collect_test.go b/search_collect_test.go new file mode 100644 index 00000000..94ef4a4e --- /dev/null +++ b/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") + } + }) +} diff --git a/search_commands.go b/search_commands.go index 935ae0a7..235df0c3 100644 --- a/search_commands.go +++ b/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 "" }