hanzokv/go
1
0
Fork 0
mirror of https://github.com/hanzokv/go.git synced 2026-09-24 00:31:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(ft.aggregate): Add Steps for query building (#3782)

Browse Source 
* ft.aggregate with steps

* ignore deprecated fields when steps are present

* add example app

* address pr comments

---------

Co-authored-by: Elena Kolevska <[email protected]>
This commit is contained in:
Nedyalko Dyakov
2026-04-25 00:21:16 +03:00
committed by GitHub
co-authored by Elena Kolevska
parent d9d7694911
commit 22b26f4b3e
8 changed files with 704 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doctests/query_agg_test.go
+20 -30
View File
@@ -228,15 +228,13 @@ func ExampleClient_query_agg() {
"idx:bicycle",
"@condition:{new}",
&redis.FTAggregateOptions{
Apply: []redis.FTAggregateApply{
{
Steps: []redis.FTAggregateStep{
{Load: &redis.FTAggregateLoad{Field: "__key"}},
{Load: &redis.FTAggregateLoad{Field: "price"}},
{Apply: &redis.FTAggregateApply{
Field: "@price - (@price * 0.1)",
As: "discounted",
},
},
Load: []redis.FTAggregateLoad{
{Field: "__key"},
{Field: "price"},
}},
},
},
).Result()
@@ -271,17 +269,13 @@ func ExampleClient_query_agg() {
res2, err := rdb.FTAggregateWithArgs(ctx,
"idx:bicycle", "*",
&redis.FTAggregateOptions{
Load: []redis.FTAggregateLoad{
{Field: "price"},
},
Apply: []redis.FTAggregateApply{
{
Steps: []redis.FTAggregateStep{
{Load: &redis.FTAggregateLoad{Field: "price"}},
{Apply: &redis.FTAggregateApply{
Field: "@price<1000",
As: "price_category",
},
},
GroupBy: []redis.FTAggregateGroupBy{
{
}},
{GroupBy: &redis.FTAggregateGroupBy{
Fields: []interface{}{"@condition"},
Reduce: []redis.FTAggregateReducer{
{
@@ -290,7 +284,7 @@ func ExampleClient_query_agg() {
As: "num_affordable",
},
},
},
}},
},
},
).Result()
@@ -323,14 +317,12 @@ func ExampleClient_query_agg() {
res3, err := rdb.FTAggregateWithArgs(ctx,
"idx:bicycle", "*",
&redis.FTAggregateOptions{
Apply: []redis.FTAggregateApply{
{
Steps: []redis.FTAggregateStep{
{Apply: &redis.FTAggregateApply{
Field: "'bicycle'",
As: "type",
},
},
GroupBy: []redis.FTAggregateGroupBy{
{
}},
{GroupBy: &redis.FTAggregateGroupBy{
Fields: []interface{}{"@type"},
Reduce: []redis.FTAggregateReducer{
{
@@ -338,7 +330,7 @@ func ExampleClient_query_agg() {
As: "num_total",
},
},
},
}},
},
},
).Result()
@@ -363,11 +355,9 @@ func ExampleClient_query_agg() {
res4, err := rdb.FTAggregateWithArgs(ctx,
"idx:bicycle", "*",
&redis.FTAggregateOptions{
Load: []redis.FTAggregateLoad{
{Field: "__key"},
},
GroupBy: []redis.FTAggregateGroupBy{
{
Steps: []redis.FTAggregateStep{
{Load: &redis.FTAggregateLoad{Field: "__key"}},
{GroupBy: &redis.FTAggregateGroupBy{
Fields: []interface{}{"@condition"},
Reduce: []redis.FTAggregateReducer{
{
@@ -376,7 +366,7 @@ func ExampleClient_query_agg() {
As: "bicycles",
},
},
},
}},
},
},
).Result()
example/search-aggregate-steps/README.md
+32
View File
@@ -0,0 +1,32 @@
# FT.AGGREGATE step-based pipeline example
This example demonstrates `FTAggregateOptions.Steps`, which lets `LOAD`,
`APPLY`, `GROUPBY` and `SORTBY` appear multiple times in any order in an
`FT.AGGREGATE` pipeline.
The main motivation is shard-level trimming: running a `SORTBY ... MAX N`
**before** a `GROUPBY` so each shard only ships its top-N rows onward.
## Prerequisites
A Redis instance with the Search module loaded (Redis 8+, or Redis Stack)
listening on `localhost:6379`.
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.
## Run
```shell
go run .
```
The example:
1. Creates an index `idx:products` over hashes prefixed with `product:`.
2. Seeds 10 products with `category`, `brand`, `price`, `quantity`, `rating`.
3. Runs three aggregations:
- **Trim before GROUPBY** — `SORTBY @rating DESC MAX 5` → `GROUPBY @category` → `SORTBY @price_total DESC`.
- **Multi-stage pipeline** — `LOAD` → `LOAD` → `APPLY` → `SORTBY MAX` → `GROUPBY` → `APPLY` → `SORTBY`.
- **AggregateBuilder** — the same first query expressed via the fluent `NewAggregateBuilder` API.
example/search-aggregate-steps/go.mod
+12
View File
@@ -0,0 +1,12 @@
module github.com/redis/go-redis/example/search-aggregate-steps
go 1.24
replace github.com/redis/go-redis/v9 => ../..
require github.com/redis/go-redis/v9 v9.18.0
require (
github.com/cespare/xxhash/v2 v2.3.0 // indirect
go.uber.org/atomic v1.11.0 // indirect
)
example/search-aggregate-steps/go.sum
+20
View File
@@ -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-steps/main.go
+150
View File
@@ -0,0 +1,150 @@
// Example demonstrating the step-based FT.AGGREGATE API.
//
// FTAggregateOptions.Steps lets LOAD, APPLY, GROUPBY and SORTBY appear
// multiple times in any order. This is useful for shard-level trimming
// (SORTBY ... MAX N) before GROUPBY and multi-stage pipelines.
//
// Requires a Redis instance with the Search module (e.g. Redis 8+).
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()
if err := rdb.FlushDB(ctx).Err(); err != nil {
log.Fatalf("flushdb: %v", err)
}
createIndex(ctx, rdb)
seedProducts(ctx, rdb)
fmt.Println("# 1. Trim before GROUPBY (real-world use case)")
trimBeforeGroup(ctx, rdb)
fmt.Println("\n# 2. Multi-stage pipeline (LOAD -> APPLY -> SORTBY -> GROUPBY -> APPLY -> SORTBY)")
multiStagePipeline(ctx, rdb)
fmt.Println("\n# 3. Same as #1 but using AggregateBuilder")
withBuilder(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: "category", FieldType: redis.SearchFieldTypeTag},
&redis.FieldSchema{FieldName: "brand", FieldType: redis.SearchFieldTypeTag},
&redis.FieldSchema{FieldName: "price", FieldType: redis.SearchFieldTypeNumeric, Sortable: true},
&redis.FieldSchema{FieldName: "quantity", 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{}{
{"category": "bike", "brand": "Velorim", "price": 270, "quantity": 12, "rating": 4.5},
{"category": "bike", "brand": "Velorim", "price": 810, "quantity": 3, "rating": 4.1},
{"category": "bike", "brand": "Bicyk", "price": 2300, "quantity": 1, "rating": 4.9},
{"category": "bike", "brand": "Bicyk", "price": 430, "quantity": 7, "rating": 3.8},
{"category": "helmet", "brand": "Nord", "price": 120, "quantity": 25, "rating": 4.7},
{"category": "helmet", "brand": "Nord", "price": 80, "quantity": 40, "rating": 4.0},
{"category": "helmet", "brand": "Peaknetic", "price": 200, "quantity": 10, "rating": 4.8},
{"category": "light", "brand": "Peaknetic", "price": 45, "quantity": 60, "rating": 4.3},
{"category": "light", "brand": "Nord", "price": 30, "quantity": 100, "rating": 3.9},
{"category": "light", "brand": "Velorim", "price": 65, "quantity": 35, "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)
}
}
}
// trimBeforeGroup shows SORTBY ... MAX N applied BEFORE a GROUPBY
// (the pattern that the old pipeline-ordered API could not express).
func trimBeforeGroup(ctx context.Context, rdb *redis.Client) {
opts := &redis.FTAggregateOptions{
Steps: []redis.FTAggregateStep{
// Keep only the top 5 highest-rated products per shard.
{SortBy: &redis.FTAggregateSortByStep{
Fields: []redis.FTAggregateSortBy{{FieldName: "@rating", Desc: true}},
Max: 5,
}},
// Sum their price per category.
{GroupBy: &redis.FTAggregateGroupBy{
Fields: []interface{}{"@category"},
Reduce: []redis.FTAggregateReducer{
{Reducer: redis.SearchSum, Args: []interface{}{"@price"}, As: "price_total"},
{Reducer: redis.SearchCount, As: "n"},
},
}},
// Sort categories by the aggregated total.
{SortBy: &redis.FTAggregateSortByStep{
Fields: []redis.FTAggregateSortBy{{FieldName: "@price_total", Desc: true}},
}},
},
}
printRows(rdb.FTAggregateWithArgs(ctx, indexName, "*", opts).Result())
}
func multiStagePipeline(ctx context.Context, rdb *redis.Client) {
opts := &redis.FTAggregateOptions{
Steps: []redis.FTAggregateStep{
{Load: &redis.FTAggregateLoad{Field: "@price"}},
{Load: &redis.FTAggregateLoad{Field: "@quantity"}},
{Apply: &redis.FTAggregateApply{Field: "@price * @quantity", As: "stock_value"}},
{SortBy: &redis.FTAggregateSortByStep{
Fields: []redis.FTAggregateSortBy{{FieldName: "@stock_value", Desc: true}},
Max: 8,
}},
{GroupBy: &redis.FTAggregateGroupBy{
Fields: []interface{}{"@brand"},
Reduce: []redis.FTAggregateReducer{
{Reducer: redis.SearchSum, Args: []interface{}{"@stock_value"}, As: "brand_value"},
},
}},
{Apply: &redis.FTAggregateApply{Field: "floor(@brand_value)", As: "brand_value"}},
{SortBy: &redis.FTAggregateSortByStep{
Fields: []redis.FTAggregateSortBy{{FieldName: "@brand_value", Desc: true}},
}},
},
}
printRows(rdb.FTAggregateWithArgs(ctx, indexName, "*", opts).Result())
}
func withBuilder(ctx context.Context, rdb *redis.Client) {
res, err := rdb.NewAggregateBuilder(ctx, indexName, "*").
SortBy("@rating", false).SortByMax(5).
GroupBy("@category").
ReduceAs(redis.SearchSum, "price_total", "@price").
ReduceAs(redis.SearchCount, "n").
SortBy("@price_total", false).
Run()
printRows(res, err)
}
func printRows(res *redis.FTAggregateResult, err error) {
if err != nil {
log.Fatalf("aggregate: %v", err)
}
for _, row := range res.Rows {
fmt.Printf(" %v\n", row.Fields)
}
}
search_builders.go
+65 -32
View File
@@ -2,6 +2,7 @@ package redis
import (
"context"
"fmt"
)
// ----------------------
@@ -215,6 +216,7 @@ type AggregateBuilder struct {
index string
query string
options *FTAggregateOptions
err error
}
// NewAggregateBuilder creates a new AggregateBuilder for FT.AGGREGATE commands.
@@ -223,6 +225,14 @@ func (c *Client) NewAggregateBuilder(ctx context.Context, index, query string) *
return &AggregateBuilder{c: c, ctx: ctx, index: index, query: query, options: &FTAggregateOptions{LimitOffset: -1}}
}
// setErr records the first error produced while building the pipeline.
// Subsequent errors are ignored; the first error is returned from Run.
func (b *AggregateBuilder) setErr(err error) {
if b.err == nil {
b.err = err
}
}
// Verbatim includes VERBATIM.
func (b *AggregateBuilder) Verbatim() *AggregateBuilder { b.options.Verbatim = true; return b }
@@ -241,15 +251,15 @@ func (b *AggregateBuilder) LoadAll() *AggregateBuilder {
return b
}
// Load adds LOAD <n> <field> [AS alias]...
// You can call it multiple times for multiple fields.
// Load adds a LOAD <field> [AS alias] step.
// You can call it multiple times; each call becomes a separate LOAD clause
// at its position in the pipeline.
func (b *AggregateBuilder) Load(field string, alias ...string) *AggregateBuilder {
// each Load entry becomes one element in options.Load
l := FTAggregateLoad{Field: field}
l := &FTAggregateLoad{Field: field}
if len(alias) > 0 {
l.As = alias[0]
}
b.options.Load = append(b.options.Load, l)
b.options.Steps = append(b.options.Steps, FTAggregateStep{Load: l})
return b
}
@@ -259,62 +269,79 @@ func (b *AggregateBuilder) Timeout(ms int) *AggregateBuilder {
return b
}
// Apply adds APPLY <field> [AS alias].
// Apply adds an APPLY <field> [AS alias] step.
func (b *AggregateBuilder) Apply(field string, alias ...string) *AggregateBuilder {
a := FTAggregateApply{Field: field}
a := &FTAggregateApply{Field: field}
if len(alias) > 0 {
a.As = alias[0]
}
b.options.Apply = append(b.options.Apply, a)
b.options.Steps = append(b.options.Steps, FTAggregateStep{Apply: a})
return b
}
// GroupBy starts a new GROUPBY <fields...> clause.
// GroupBy adds a new GROUPBY <fields...> step.
func (b *AggregateBuilder) GroupBy(fields ...interface{}) *AggregateBuilder {
b.options.GroupBy = append(b.options.GroupBy, FTAggregateGroupBy{
Fields: fields,
b.options.Steps = append(b.options.Steps, FTAggregateStep{
GroupBy: &FTAggregateGroupBy{Fields: fields},
})
return b
}
// Reduce adds a REDUCE <fn> [<#args> <args...>] clause to the *last* GROUPBY.
// Reduce adds a REDUCE <fn> [<#args> <args...>] clause to the last step,
// which must be a GROUPBY. If it is not, Run will return an error.
func (b *AggregateBuilder) Reduce(fn SearchAggregator, args ...interface{}) *AggregateBuilder {
if len(b.options.GroupBy) == 0 {
// no GROUPBY yet — nothing to attach to
n := len(b.options.Steps)
if n == 0 || b.options.Steps[n-1].GroupBy == nil {
b.setErr(fmt.Errorf("FT.AGGREGATE: Reduce must follow a GroupBy step"))
return b
}
idx := len(b.options.GroupBy) - 1
b.options.GroupBy[idx].Reduce = append(b.options.GroupBy[idx].Reduce, FTAggregateReducer{
Reducer: fn,
Args: args,
})
g := b.options.Steps[n-1].GroupBy
g.Reduce = append(g.Reduce, FTAggregateReducer{Reducer: fn, Args: args})
return b
}
// ReduceAs does the same but also sets an alias: REDUCE <fn> … AS <alias>
// ReduceAs does the same but also sets an alias: REDUCE <fn> … AS <alias>.
// The last step must be a GROUPBY; otherwise Run will return an error.
func (b *AggregateBuilder) ReduceAs(fn SearchAggregator, alias string, args ...interface{}) *AggregateBuilder {
if len(b.options.GroupBy) == 0 {
n := len(b.options.Steps)
if n == 0 || b.options.Steps[n-1].GroupBy == nil {
b.setErr(fmt.Errorf("FT.AGGREGATE: ReduceAs must follow a GroupBy step"))
return b
}
idx := len(b.options.GroupBy) - 1
b.options.GroupBy[idx].Reduce = append(b.options.GroupBy[idx].Reduce, FTAggregateReducer{
Reducer: fn,
Args: args,
As: alias,
})
g := b.options.Steps[n-1].GroupBy
g.Reduce = append(g.Reduce, FTAggregateReducer{Reducer: fn, Args: args, As: alias})
return b
}
// SortBy adds SORTBY <field> ASC|DESC.
// 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
// SORTBY step.
//
// Note: this is a semantics change from earlier experimental versions of
// the builder, where SortBy always accumulated into a single SORTBY clause
// regardless of position in the pipeline.
func (b *AggregateBuilder) SortBy(field string, asc bool) *AggregateBuilder {
sb := FTAggregateSortBy{FieldName: field, Asc: asc, Desc: !asc}
b.options.SortBy = append(b.options.SortBy, sb)
if n := len(b.options.Steps); n > 0 && b.options.Steps[n-1].SortBy != nil {
b.options.Steps[n-1].SortBy.Fields = append(b.options.Steps[n-1].SortBy.Fields, sb)
return b
}
b.options.Steps = append(b.options.Steps, FTAggregateStep{
SortBy: &FTAggregateSortByStep{Fields: []FTAggregateSortBy{sb}},
})
return b
}
// SortByMax sets MAX <n> (only if SortBy was called).
// SortByMax sets MAX <n> on the last SORTBY step. The last step must be a
// SORTBY; otherwise Run will return an error.
func (b *AggregateBuilder) SortByMax(max int) *AggregateBuilder {
b.options.SortByMax = max
n := len(b.options.Steps)
if n == 0 || b.options.Steps[n-1].SortBy == nil {
b.setErr(fmt.Errorf("FT.AGGREGATE: SortByMax must follow a SortBy step"))
return b
}
b.options.Steps[n-1].SortBy.Max = max
return b
}
@@ -352,8 +379,14 @@ func (b *AggregateBuilder) Dialect(version int) *AggregateBuilder {
return b
}
// Run executes FT.AGGREGATE and returns a typed result.
// Run executes FT.AGGREGATE and returns a typed result. If the builder
// recorded a validation error while constructing the pipeline (for example,
// calling SortByMax when the last step is not a SortBy), that error is
// returned without issuing the command.
func (b *AggregateBuilder) Run() (*FTAggregateResult, error) {
if b.err != nil {
return nil, b.err
}
cmd := b.c.FTAggregateWithArgs(b.ctx, b.index, b.query, b.options)
return cmd.Result()
}
search_commands.go
+254 -101
View File
@@ -258,22 +258,42 @@ type FTAggregateWithCursor struct {
MaxIdle int
}
// FTAggregateSortByStep represents a SORTBY operation with optional MAX.
// Used inside FTAggregateStep to place SORTBY at an arbitrary position in
// the aggregation pipeline.
type FTAggregateSortByStep struct {
Fields []FTAggregateSortBy
Max int // 0 means no MAX
}
// FTAggregateStep represents a single operation in the aggregation pipeline.
// LOAD, APPLY, SORTBY and GROUPBY can all appear multiple times in any order.
// Exactly one of the fields should be set per step.
type FTAggregateStep struct {
Load *FTAggregateLoad
Apply *FTAggregateApply
GroupBy *FTAggregateGroupBy
SortBy *FTAggregateSortByStep
}
type FTAggregateOptions struct {
Verbatim bool
LoadAll bool
Load []FTAggregateLoad
Timeout int
GroupBy []FTAggregateGroupBy
SortBy []FTAggregateSortBy
SortByMax int
Verbatim bool
LoadAll bool
Timeout int
// Scorer is used to set scoring function, if not set passed, a default will be used.
// The default scorer depends on the Redis version:
// - `BM25` for Redis >= 8
// - `TFIDF` for Redis < 8
Scorer string
// AddScores is available in Redis CE 8
AddScores bool
Apply []FTAggregateApply
AddScores bool
// Steps is the ordered sequence of aggregation pipeline operations.
// It can contain LOAD, APPLY, GROUPBY and SORTBY in any order, multiple times.
// Steps cannot be combined with the deprecated Load, Apply, GroupBy, SortBy
// and SortByMax fields: doing so returns an error.
Steps []FTAggregateStep
LimitOffset int
Limit int
Filter string
@@ -282,6 +302,17 @@ type FTAggregateOptions struct {
Params map[string]interface{}
// Dialect 1,3 and 4 are deprecated since redis 8.0
DialectVersion int
// Deprecated: Use Steps instead.
Load []FTAggregateLoad
// Deprecated: Use Steps instead.
GroupBy []FTAggregateGroupBy
// Deprecated: Use Steps instead.
SortBy []FTAggregateSortBy
// Deprecated: Use Steps instead.
SortByMax int
// Deprecated: Use Steps instead.
Apply []FTAggregateApply
}
type FTSearchFilter struct {
@@ -617,9 +648,112 @@ func (c cmdable) FTAggregate(ctx context.Context, index string, query string) *M
return cmd
}
// validateFTAggregateOptions validates mutually exclusive combinations of
// FTAggregateOptions fields before any command arguments are constructed.
func validateFTAggregateOptions(options *FTAggregateOptions) error {
if len(options.Steps) > 0 {
if options.Load != nil || options.Apply != nil || options.GroupBy != nil ||
options.SortBy != nil || options.SortByMax != 0 {
return fmt.Errorf("FT.AGGREGATE: Steps cannot be combined with the deprecated Load, Apply, GroupBy, SortBy and SortByMax fields")
}
if options.LoadAll {
for _, step := range options.Steps {
if step.Load != nil {
return fmt.Errorf("FT.AGGREGATE: LOADALL and LOAD are mutually exclusive")
}
}
}
}
if options.LoadAll && options.Load != nil {
return fmt.Errorf("FT.AGGREGATE: LOADALL and LOAD are mutually exclusive")
}
return nil
}
// appendFTAggregateStep appends the Redis command arguments for a single
// aggregation pipeline step. Each step must set exactly one of Load, Apply,
// GroupBy or SortBy.
func appendFTAggregateStep(args []interface{}, step FTAggregateStep) ([]interface{}, error) {
set := 0
if step.Load != nil {
set++
}
if step.Apply != nil {
set++
}
if step.GroupBy != nil {
set++
}
if step.SortBy != nil {
set++
}
if set != 1 {
return args, fmt.Errorf("FT.AGGREGATE: each step must set exactly one of Load, Apply, GroupBy, SortBy (got %d)", set)
}
switch {
case step.Load != nil:
args = append(args, "LOAD")
countIdx := len(args)
args = append(args, 0)
count := 0
args = append(args, step.Load.Field)
count++
if step.Load.As != "" {
args = append(args, "AS", step.Load.As)
count += 2
}
args[countIdx] = count
case step.Apply != nil:
args = append(args, "APPLY", step.Apply.Field)
if step.Apply.As != "" {
args = append(args, "AS", step.Apply.As)
}
case step.GroupBy != nil:
args = append(args, "GROUPBY", len(step.GroupBy.Fields))
args = append(args, step.GroupBy.Fields...)
for _, reducer := range step.GroupBy.Reduce {
args = append(args, "REDUCE", reducer.Reducer.String())
if reducer.Args != nil {
args = append(args, len(reducer.Args))
args = append(args, reducer.Args...)
} else {
args = append(args, 0)
}
if reducer.As != "" {
args = append(args, "AS", reducer.As)
}
}
case step.SortBy != nil:
args = append(args, "SORTBY")
sortByOptions := []interface{}{}
for _, sortBy := range step.SortBy.Fields {
if sortBy.Asc && sortBy.Desc {
return args, fmt.Errorf("FT.AGGREGATE: ASC and DESC are mutually exclusive")
}
sortByOptions = append(sortByOptions, sortBy.FieldName)
if sortBy.Asc {
sortByOptions = append(sortByOptions, "ASC")
}
if sortBy.Desc {
sortByOptions = append(sortByOptions, "DESC")
}
}
args = append(args, len(sortByOptions))
args = append(args, sortByOptions...)
if step.SortBy.Max > 0 {
args = append(args, "MAX", step.SortBy.Max)
}
}
return args, nil
}
func FTAggregateQuery(query string, options *FTAggregateOptions) (AggregateQuery, error) {
queryArgs := []interface{}{query}
if options != nil {
if err := validateFTAggregateOptions(options); err != nil {
return nil, err
}
if options.Verbatim {
queryArgs = append(queryArgs, "VERBATIM")
}
@@ -632,13 +766,10 @@ func FTAggregateQuery(query string, options *FTAggregateOptions) (AggregateQuery
queryArgs = append(queryArgs, "ADDSCORES")
}
if options.LoadAll && options.Load != nil {
return nil, fmt.Errorf("FT.AGGREGATE: LOADALL and LOAD are mutually exclusive")
}
if options.LoadAll {
queryArgs = append(queryArgs, "LOAD", "*")
}
if options.Load != nil {
if len(options.Steps) == 0 && options.Load != nil {
queryArgs = append(queryArgs, "LOAD", len(options.Load))
index, count := len(queryArgs)-1, 0
for _, load := range options.Load {
@@ -656,53 +787,63 @@ func FTAggregateQuery(query string, options *FTAggregateOptions) (AggregateQuery
queryArgs = append(queryArgs, "TIMEOUT", options.Timeout)
}
for _, apply := range options.Apply {
queryArgs = append(queryArgs, "APPLY", apply.Field)
if apply.As != "" {
queryArgs = append(queryArgs, "AS", apply.As)
if len(options.Steps) > 0 {
for _, step := range options.Steps {
var err error
queryArgs, err = appendFTAggregateStep(queryArgs, step)
if err != nil {
return nil, err
}
}
} else {
for _, apply := range options.Apply {
queryArgs = append(queryArgs, "APPLY", apply.Field)
if apply.As != "" {
queryArgs = append(queryArgs, "AS", apply.As)
}
}
}
if options.GroupBy != nil {
for _, groupBy := range options.GroupBy {
queryArgs = append(queryArgs, "GROUPBY", len(groupBy.Fields))
queryArgs = append(queryArgs, groupBy.Fields...)
if options.GroupBy != nil {
for _, groupBy := range options.GroupBy {
queryArgs = append(queryArgs, "GROUPBY", len(groupBy.Fields))
queryArgs = append(queryArgs, groupBy.Fields...)
for _, reducer := range groupBy.Reduce {
queryArgs = append(queryArgs, "REDUCE")
queryArgs = append(queryArgs, reducer.Reducer.String())
if reducer.Args != nil {
queryArgs = append(queryArgs, len(reducer.Args))
queryArgs = append(queryArgs, reducer.Args...)
} else {
queryArgs = append(queryArgs, 0)
}
if reducer.As != "" {
queryArgs = append(queryArgs, "AS", reducer.As)
for _, reducer := range groupBy.Reduce {
queryArgs = append(queryArgs, "REDUCE")
queryArgs = append(queryArgs, reducer.Reducer.String())
if reducer.Args != nil {
queryArgs = append(queryArgs, len(reducer.Args))
queryArgs = append(queryArgs, reducer.Args...)
} else {
queryArgs = append(queryArgs, 0)
}
if reducer.As != "" {
queryArgs = append(queryArgs, "AS", reducer.As)
}
}
}
}
}
if options.SortBy != nil {
queryArgs = append(queryArgs, "SORTBY")
sortByOptions := []interface{}{}
for _, sortBy := range options.SortBy {
sortByOptions = append(sortByOptions, sortBy.FieldName)
if sortBy.Asc && sortBy.Desc {
return nil, fmt.Errorf("FT.AGGREGATE: ASC and DESC are mutually exclusive")
}
if sortBy.Asc {
sortByOptions = append(sortByOptions, "ASC")
}
if sortBy.Desc {
sortByOptions = append(sortByOptions, "DESC")
if options.SortBy != nil {
queryArgs = append(queryArgs, "SORTBY")
sortByOptions := []interface{}{}
for _, sortBy := range options.SortBy {
sortByOptions = append(sortByOptions, sortBy.FieldName)
if sortBy.Asc && sortBy.Desc {
return nil, fmt.Errorf("FT.AGGREGATE: ASC and DESC are mutually exclusive")
}
if sortBy.Asc {
sortByOptions = append(sortByOptions, "ASC")
}
if sortBy.Desc {
sortByOptions = append(sortByOptions, "DESC")
}
}
queryArgs = append(queryArgs, len(sortByOptions))
queryArgs = append(queryArgs, sortByOptions...)
}
if options.SortByMax > 0 {
queryArgs = append(queryArgs, "MAX", options.SortByMax)
}
queryArgs = append(queryArgs, len(sortByOptions))
queryArgs = append(queryArgs, sortByOptions...)
}
if options.SortByMax > 0 {
queryArgs = append(queryArgs, "MAX", options.SortByMax)
}
if options.LimitOffset >= 0 && options.Limit > 0 {
queryArgs = append(queryArgs, "LIMIT", options.LimitOffset, options.Limit)
@@ -852,6 +993,11 @@ func (cmd *AggregateCmd) Clone() Cmder {
func (c cmdable) FTAggregateWithArgs(ctx context.Context, index string, query string, options *FTAggregateOptions) *AggregateCmd {
args := []interface{}{"FT.AGGREGATE", index, query}
if options != nil {
if err := validateFTAggregateOptions(options); err != nil {
cmd := NewAggregateCmd(ctx, args...)
cmd.SetErr(err)
return cmd
}
if options.Verbatim {
args = append(args, "VERBATIM")
}
@@ -861,15 +1007,10 @@ func (c cmdable) FTAggregateWithArgs(ctx context.Context, index string, query st
if options.AddScores {
args = append(args, "ADDSCORES")
}
if options.LoadAll && options.Load != nil {
cmd := NewAggregateCmd(ctx, args...)
cmd.SetErr(fmt.Errorf("FT.AGGREGATE: LOADALL and LOAD are mutually exclusive"))
return cmd
}
if options.LoadAll {
args = append(args, "LOAD", "*")
}
if options.Load != nil {
if len(options.Steps) == 0 && options.Load != nil {
args = append(args, "LOAD", len(options.Load))
index, count := len(args)-1, 0
for _, load := range options.Load {
@@ -885,54 +1026,66 @@ func (c cmdable) FTAggregateWithArgs(ctx context.Context, index string, query st
if options.Timeout > 0 {
args = append(args, "TIMEOUT", options.Timeout)
}
for _, apply := range options.Apply {
args = append(args, "APPLY", apply.Field)
if apply.As != "" {
args = append(args, "AS", apply.As)
}
}
if options.GroupBy != nil {
for _, groupBy := range options.GroupBy {
args = append(args, "GROUPBY", len(groupBy.Fields))
args = append(args, groupBy.Fields...)
for _, reducer := range groupBy.Reduce {
args = append(args, "REDUCE")
args = append(args, reducer.Reducer.String())
if reducer.Args != nil {
args = append(args, len(reducer.Args))
args = append(args, reducer.Args...)
} else {
args = append(args, 0)
}
if reducer.As != "" {
args = append(args, "AS", reducer.As)
}
}
}
}
if options.SortBy != nil {
args = append(args, "SORTBY")
sortByOptions := []interface{}{}
for _, sortBy := range options.SortBy {
sortByOptions = append(sortByOptions, sortBy.FieldName)
if sortBy.Asc && sortBy.Desc {
if len(options.Steps) > 0 {
for _, step := range options.Steps {
var err error
args, err = appendFTAggregateStep(args, step)
if err != nil {
cmd := NewAggregateCmd(ctx, args...)
cmd.SetErr(fmt.Errorf("FT.AGGREGATE: ASC and DESC are mutually exclusive"))
cmd.SetErr(err)
return cmd
}
if sortBy.Asc {
sortByOptions = append(sortByOptions, "ASC")
}
if sortBy.Desc {
sortByOptions = append(sortByOptions, "DESC")
}
} else {
for _, apply := range options.Apply {
args = append(args, "APPLY", apply.Field)
if apply.As != "" {
args = append(args, "AS", apply.As)
}
}
args = append(args, len(sortByOptions))
args = append(args, sortByOptions...)
}
if options.SortByMax > 0 {
args = append(args, "MAX", options.SortByMax)
if options.GroupBy != nil {
for _, groupBy := range options.GroupBy {
args = append(args, "GROUPBY", len(groupBy.Fields))
args = append(args, groupBy.Fields...)
for _, reducer := range groupBy.Reduce {
args = append(args, "REDUCE")
args = append(args, reducer.Reducer.String())
if reducer.Args != nil {
args = append(args, len(reducer.Args))
args = append(args, reducer.Args...)
} else {
args = append(args, 0)
}
if reducer.As != "" {
args = append(args, "AS", reducer.As)
}
}
}
}
if options.SortBy != nil {
args = append(args, "SORTBY")
sortByOptions := []interface{}{}
for _, sortBy := range options.SortBy {
sortByOptions = append(sortByOptions, sortBy.FieldName)
if sortBy.Asc && sortBy.Desc {
cmd := NewAggregateCmd(ctx, args...)
cmd.SetErr(fmt.Errorf("FT.AGGREGATE: ASC and DESC are mutually exclusive"))
return cmd
}
if sortBy.Asc {
sortByOptions = append(sortByOptions, "ASC")
}
if sortBy.Desc {
sortByOptions = append(sortByOptions, "DESC")
}
}
args = append(args, len(sortByOptions))
args = append(args, sortByOptions...)
}
if options.SortByMax > 0 {
args = append(args, "MAX", options.SortByMax)
}
}
if options.LimitOffset >= 0 && options.Limit > 0 {
args = append(args, "LIMIT", options.LimitOffset, options.Limit)
search_test.go
+151
View File
@@ -902,6 +902,157 @@ var _ = Describe("RediSearch commands Resp 2", Label("search"), func() {
Expect(found).To(BeTrue())
})
It("should emit Steps in order with SORTBY MAX and GROUPBY", Label("search", "ftaggregate"), func() {
options := &redis.FTAggregateOptions{
Scorer: "BM25STD.TANH",
AddScores: true,
Steps: []redis.FTAggregateStep{
{SortBy: &redis.FTAggregateSortByStep{
Fields: []redis.FTAggregateSortBy{{FieldName: "@__score", Desc: true}},
Max: 50000,
}},
{GroupBy: &redis.FTAggregateGroupBy{
Fields: []interface{}{"@parent_id", "@profile_id"},
Reduce: []redis.FTAggregateReducer{
{Reducer: redis.SearchSum, Args: []interface{}{"the_score"}, As: "the_score"},
},
}},
{SortBy: &redis.FTAggregateSortByStep{
Fields: []redis.FTAggregateSortBy{{FieldName: "@the_score", Desc: true}},
}},
},
LimitOffset: 0,
Limit: 400,
}
args, err := redis.FTAggregateQuery("q", options)
Expect(err).NotTo(HaveOccurred())
expected := redis.AggregateQuery{
"q",
"SCORER", "BM25STD.TANH",
"ADDSCORES",
"SORTBY", 2, "@__score", "DESC", "MAX", 50000,
"GROUPBY", 2, "@parent_id", "@profile_id",
"REDUCE", "SUM", 1, "the_score", "AS", "the_score",
"SORTBY", 2, "@the_score", "DESC",
"LIMIT", 0, 400,
"DIALECT", 2,
}
Expect(args).To(Equal(expected))
})
It("should emit Steps with LOAD and APPLY interleaved", Label("search", "ftaggregate"), func() {
options := &redis.FTAggregateOptions{
Steps: []redis.FTAggregateStep{
{Load: &redis.FTAggregateLoad{Field: "@price", As: "p"}},
{Load: &redis.FTAggregateLoad{Field: "@quantity"}},
{Apply: &redis.FTAggregateApply{Field: "@p * @quantity", As: "total"}},
{SortBy: &redis.FTAggregateSortByStep{
Fields: []redis.FTAggregateSortBy{{FieldName: "@total", Desc: true}},
Max: 10000,
}},
},
}
args, err := redis.FTAggregateQuery("*", options)
Expect(err).NotTo(HaveOccurred())
expected := redis.AggregateQuery{
"*",
"LOAD", 3, "@price", "AS", "p",
"LOAD", 1, "@quantity",
"APPLY", "@p * @quantity", "AS", "total",
"SORTBY", 2, "@total", "DESC", "MAX", 10000,
"DIALECT", 2,
}
Expect(args).To(Equal(expected))
})
It("should error when LoadAll is combined with a LOAD step", Label("search", "ftaggregate"), func() {
options := &redis.FTAggregateOptions{
LoadAll: true,
Steps: []redis.FTAggregateStep{
{Load: &redis.FTAggregateLoad{Field: "@price"}},
{Apply: &redis.FTAggregateApply{Field: "@price * 2", As: "double"}},
},
}
_, err := redis.FTAggregateQuery("*", options)
Expect(err).To(MatchError(ContainSubstring("LOADALL and LOAD are mutually exclusive")))
})
It("should allow LoadAll together with Steps that don't use LOAD", Label("search", "ftaggregate"), func() {
options := &redis.FTAggregateOptions{
LoadAll: true,
Steps: []redis.FTAggregateStep{
{Apply: &redis.FTAggregateApply{Field: "@price * 2", As: "double"}},
},
}
args, err := redis.FTAggregateQuery("*", options)
Expect(err).NotTo(HaveOccurred())
expected := redis.AggregateQuery{
"*",
"LOAD", "*",
"APPLY", "@price * 2", "AS", "double",
"DIALECT", 2,
}
Expect(args).To(Equal(expected))
})
It("should error when SORTBY step has both ASC and DESC", Label("search", "ftaggregate"), func() {
options := &redis.FTAggregateOptions{
Steps: []redis.FTAggregateStep{
{SortBy: &redis.FTAggregateSortByStep{
Fields: []redis.FTAggregateSortBy{{FieldName: "@x", Asc: true, Desc: true}},
}},
},
}
_, err := redis.FTAggregateQuery("q", options)
Expect(err).To(HaveOccurred())
})
It("should error when a step sets more than one field", Label("search", "ftaggregate"), func() {
options := &redis.FTAggregateOptions{
Steps: []redis.FTAggregateStep{
{
Load: &redis.FTAggregateLoad{Field: "@x"},
Apply: &redis.FTAggregateApply{Field: "@x * 2", As: "y"},
},
},
}
_, err := redis.FTAggregateQuery("q", options)
Expect(err).To(MatchError(ContainSubstring("each step must set exactly one")))
})
It("should error when a step sets no field", Label("search", "ftaggregate"), func() {
options := &redis.FTAggregateOptions{
Steps: []redis.FTAggregateStep{{}},
}
_, err := redis.FTAggregateQuery("q", options)
Expect(err).To(MatchError(ContainSubstring("each step must set exactly one")))
})
It("should error when Steps is combined with deprecated fields", Label("search", "ftaggregate"), func() {
options := &redis.FTAggregateOptions{
Load: []redis.FTAggregateLoad{{Field: "@ignored_load"}},
Steps: []redis.FTAggregateStep{
{Apply: &redis.FTAggregateApply{Field: "@used", As: "u"}},
},
}
_, err := redis.FTAggregateQuery("q", options)
Expect(err).To(MatchError(ContainSubstring("Steps cannot be combined with the deprecated")))
})
It("should error when Steps is combined with deprecated SortByMax", Label("search", "ftaggregate"), func() {
options := &redis.FTAggregateOptions{
SortByMax: 999,
Steps: []redis.FTAggregateStep{
{Apply: &redis.FTAggregateApply{Field: "@used", As: "u"}},
},
}
_, err := redis.FTAggregateQuery("q", options)
Expect(err).To(MatchError(ContainSubstring("Steps cannot be combined with the deprecated")))
})
It("should FTSearch SkipInitialScan", Label("search", "ftsearch"), func() {
client.HSet(ctx, "doc1", "foo", "bar")