fix(aggregate, search): ft.aggregate bugfixes (#3263)

* fix: rearange args for ft.aggregate

apply should be before any groupby or sortby

* improve test

* wip: add scorer and addscores

* enable all tests

* fix ftsearch with count test

* make linter happy

* Addscores is available in later redisearch releases.

For safety state it is available in redis ce 8

* load an apply seem to break scorer and addscores

search_commands.go

@@ -247,6 +247,13 @@ type FTAggregateOptions struct {
 	GroupBy   []FTAggregateGroupBy
 	SortBy    []FTAggregateSortBy
 	SortByMax 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
 	LimitOffset       int
 	Limit             int
@@ -490,6 +497,15 @@ func FTAggregateQuery(query string, options *FTAggregateOptions) AggregateQuery
 		if options.Verbatim {
 			queryArgs = append(queryArgs, "VERBATIM")
 		}
+
+		if options.Scorer != "" {
+			queryArgs = append(queryArgs, "SCORER", options.Scorer)
+		}
+
+		if options.AddScores {
+			queryArgs = append(queryArgs, "ADDSCORES")
+		}
+
 		if options.LoadAll && options.Load != nil {
 			panic("FT.AGGREGATE: LOADALL and LOAD are mutually exclusive")
 		}
@@ -505,9 +521,18 @@ func FTAggregateQuery(query string, options *FTAggregateOptions) AggregateQuery
 				}
 			}
 		}
+
 		if options.Timeout > 0 {
 			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 options.GroupBy != nil {
 			for _, groupBy := range options.GroupBy {
 				queryArgs = append(queryArgs, "GROUPBY", len(groupBy.Fields))
@@ -549,12 +574,6 @@ func FTAggregateQuery(query string, options *FTAggregateOptions) AggregateQuery
 		if options.SortByMax > 0 {
 			queryArgs = append(queryArgs, "MAX", options.SortByMax)
 		}
-		for _, apply := range options.Apply {
-			queryArgs = append(queryArgs, "APPLY", apply.Field)
-			if apply.As != "" {
-				queryArgs = append(queryArgs, "AS", apply.As)
-			}
-		}
 		if options.LimitOffset > 0 {
 			queryArgs = append(queryArgs, "LIMIT", options.LimitOffset)
 		}
@@ -581,6 +600,7 @@ func FTAggregateQuery(query string, options *FTAggregateOptions) AggregateQuery
 				queryArgs = append(queryArgs, key, value)
 			}
 		}
+
 		if options.DialectVersion > 0 {
 			queryArgs = append(queryArgs, "DIALECT", options.DialectVersion)
 		}
@@ -661,11 +681,12 @@ func (cmd *AggregateCmd) readReply(rd *proto.Reader) (err error) {
 	data, err := rd.ReadSlice()
 	if err != nil {
 		cmd.err = err
-		return nil
+		return err
 	}
 	cmd.val, err = ProcessAggregateResult(data)
 	if err != nil {
 		cmd.err = err
+		return err
 	}
 	return nil
 }
@@ -681,6 +702,12 @@ func (c cmdable) FTAggregateWithArgs(ctx context.Context, index string, query st
 		if options.Verbatim {
 			args = append(args, "VERBATIM")
 		}
+		if options.Scorer != "" {
+			args = append(args, "SCORER", options.Scorer)
+		}
+		if options.AddScores {
+			args = append(args, "ADDSCORES")
+		}
 		if options.LoadAll && options.Load != nil {
 			panic("FT.AGGREGATE: LOADALL and LOAD are mutually exclusive")
 		}
@@ -699,6 +726,12 @@ 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))
@@ -740,12 +773,6 @@ func (c cmdable) FTAggregateWithArgs(ctx context.Context, index string, query st
 		if options.SortByMax > 0 {
 			args = append(args, "MAX", options.SortByMax)
 		}
-		for _, apply := range options.Apply {
-			args = append(args, "APPLY", apply.Field)
-			if apply.As != "" {
-				args = append(args, "AS", apply.As)
-			}
-		}
 		if options.LimitOffset > 0 {
 			args = append(args, "LIMIT", options.LimitOffset)
 		}
@@ -1693,7 +1720,8 @@ func (cmd *FTSearchCmd) readReply(rd *proto.Reader) (err error) {
 
 // FTSearch - Executes a search query on an index.
 // The 'index' parameter specifies the index to search, and the 'query' parameter specifies the search query.
-// For more information, please refer to the Redis documentation:
+// For more information, please refer to the Redis documentation about [FT.SEARCH].
+//
 // [FT.SEARCH]: (https://redis.io/commands/ft.search/)
 func (c cmdable) FTSearch(ctx context.Context, index string, query string) *FTSearchCmd {
 	args := []interface{}{"FT.SEARCH", index, query}
@@ -1704,6 +1732,12 @@ func (c cmdable) FTSearch(ctx context.Context, index string, query string) *FTSe
 
 type SearchQuery []interface{}
 
+// FTSearchQuery - Executes a search query on an index with additional options.
+// The 'index' parameter specifies the index to search, the 'query' parameter specifies the search query,
+// and the 'options' parameter specifies additional options for the search.
+// For more information, please refer to the Redis documentation about [FT.SEARCH].
+//
+// [FT.SEARCH]: (https://redis.io/commands/ft.search/)
 func FTSearchQuery(query string, options *FTSearchOptions) SearchQuery {
 	queryArgs := []interface{}{query}
 	if options != nil {
@@ -1816,7 +1850,8 @@ func FTSearchQuery(query string, options *FTSearchOptions) SearchQuery {
 // FTSearchWithArgs - Executes a search query on an index with additional options.
 // The 'index' parameter specifies the index to search, the 'query' parameter specifies the search query,
 // and the 'options' parameter specifies additional options for the search.
-// For more information, please refer to the Redis documentation:
+// For more information, please refer to the Redis documentation about [FT.SEARCH].
+//
 // [FT.SEARCH]: (https://redis.io/commands/ft.search/)
 func (c cmdable) FTSearchWithArgs(ctx context.Context, index string, query string, options *FTSearchOptions) *FTSearchCmd {
 	args := []interface{}{"FT.SEARCH", index, query}
@@ -1908,7 +1943,7 @@ func (c cmdable) FTSearchWithArgs(ctx context.Context, index string, query strin
 				}
 			}
 			if options.SortByWithCount {
-				args = append(args, "WITHCOUT")
+				args = append(args, "WITHCOUNT")
 			}
 		}
 		if options.LimitOffset >= 0 && options.Limit > 0 {

search_test.go

@@ -2,6 +2,8 @@ package redis_test
 
 import (
 	"context"
+	"fmt"
+	"strconv"
 	"time"
 
 	. "github.com/bsm/ginkgo/v2"
@@ -127,8 +129,11 @@ var _ = Describe("RediSearch commands Resp 2", Label("search"), func() {
 
 		res3, err := client.FTSearchWithArgs(ctx, "num", "foo", &redis.FTSearchOptions{NoContent: true, SortBy: []redis.FTSearchSortBy{sortBy2}, SortByWithCount: true}).Result()
 		Expect(err).NotTo(HaveOccurred())
-		Expect(res3.Total).To(BeEquivalentTo(int64(0)))
+		Expect(res3.Total).To(BeEquivalentTo(int64(3)))
 
+		res4, err := client.FTSearchWithArgs(ctx, "num", "notpresentf00", &redis.FTSearchOptions{NoContent: true, SortBy: []redis.FTSearchSortBy{sortBy2}, SortByWithCount: true}).Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(res4.Total).To(BeEquivalentTo(int64(0)))
 	})
 
 	It("should FTCreate and FTSearch example", Label("search", "ftcreate", "ftsearch"), func() {
@@ -640,6 +645,100 @@ var _ = Describe("RediSearch commands Resp 2", Label("search"), func() {
 		Expect(res.Rows[0].Fields["t2"]).To(BeEquivalentTo("world"))
 	})
 
+	It("should FTAggregate with scorer and addscores", Label("search", "ftaggregate", "NonRedisEnterprise"), func() {
+		SkipBeforeRedisMajor(8, "ADDSCORES is available in Redis CE 8")
+		title := &redis.FieldSchema{FieldName: "title", FieldType: redis.SearchFieldTypeText, Sortable: false}
+		description := &redis.FieldSchema{FieldName: "description", FieldType: redis.SearchFieldTypeText, Sortable: false}
+		val, err := client.FTCreate(ctx, "idx1", &redis.FTCreateOptions{OnHash: true, Prefix: []interface{}{"product:"}}, title, description).Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(val).To(BeEquivalentTo("OK"))
+		WaitForIndexing(client, "idx1")
+
+		client.HSet(ctx, "product:1", "title", "New Gaming Laptop", "description", "this is not a desktop")
+		client.HSet(ctx, "product:2", "title", "Super Old Not Gaming Laptop", "description", "this laptop is not a new laptop but it is a laptop")
+		client.HSet(ctx, "product:3", "title", "Office PC", "description", "office desktop pc")
+
+		options := &redis.FTAggregateOptions{
+			AddScores: true,
+			Scorer:    "BM25",
+			SortBy: []redis.FTAggregateSortBy{{
+				FieldName: "@__score",
+				Desc:      true,
+			}},
+		}
+
+		res, err := client.FTAggregateWithArgs(ctx, "idx1", "laptop", options).Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(res).ToNot(BeNil())
+		Expect(len(res.Rows)).To(BeEquivalentTo(2))
+		score1, err := strconv.ParseFloat(fmt.Sprintf("%s", res.Rows[0].Fields["__score"]), 64)
+		Expect(err).NotTo(HaveOccurred())
+		score2, err := strconv.ParseFloat(fmt.Sprintf("%s", res.Rows[1].Fields["__score"]), 64)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(score1).To(BeNumerically(">", score2))
+
+		optionsDM := &redis.FTAggregateOptions{
+			AddScores: true,
+			Scorer:    "DISMAX",
+			SortBy: []redis.FTAggregateSortBy{{
+				FieldName: "@__score",
+				Desc:      true,
+			}},
+		}
+
+		resDM, err := client.FTAggregateWithArgs(ctx, "idx1", "laptop", optionsDM).Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(resDM).ToNot(BeNil())
+		Expect(len(resDM.Rows)).To(BeEquivalentTo(2))
+		score1DM, err := strconv.ParseFloat(fmt.Sprintf("%s", resDM.Rows[0].Fields["__score"]), 64)
+		Expect(err).NotTo(HaveOccurred())
+		score2DM, err := strconv.ParseFloat(fmt.Sprintf("%s", resDM.Rows[1].Fields["__score"]), 64)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(score1DM).To(BeNumerically(">", score2DM))
+
+		Expect(score1DM).To(BeEquivalentTo(float64(4)))
+		Expect(score2DM).To(BeEquivalentTo(float64(1)))
+		Expect(score1).NotTo(BeEquivalentTo(score1DM))
+		Expect(score2).NotTo(BeEquivalentTo(score2DM))
+	})
+
+	It("should FTAggregate apply and groupby", Label("search", "ftaggregate"), func() {
+		text1 := &redis.FieldSchema{FieldName: "PrimaryKey", FieldType: redis.SearchFieldTypeText, Sortable: true}
+		num1 := &redis.FieldSchema{FieldName: "CreatedDateTimeUTC", FieldType: redis.SearchFieldTypeNumeric, Sortable: true}
+		val, err := client.FTCreate(ctx, "idx1", &redis.FTCreateOptions{}, text1, num1).Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(val).To(BeEquivalentTo("OK"))
+		WaitForIndexing(client, "idx1")
+
+		// 6 feb
+		client.HSet(ctx, "doc1", "PrimaryKey", "9::362330", "CreatedDateTimeUTC", "1738823999")
+
+		// 12 feb
+		client.HSet(ctx, "doc2", "PrimaryKey", "9::362329", "CreatedDateTimeUTC", "1739342399")
+		client.HSet(ctx, "doc3", "PrimaryKey", "9::362329", "CreatedDateTimeUTC", "1739353199")
+
+		reducer := redis.FTAggregateReducer{Reducer: redis.SearchCount, As: "perDay"}
+
+		options := &redis.FTAggregateOptions{
+			Apply: []redis.FTAggregateApply{{Field: "floor(@CreatedDateTimeUTC /(60*60*24))", As: "TimestampAsDay"}},
+			GroupBy: []redis.FTAggregateGroupBy{{
+				Fields: []interface{}{"@TimestampAsDay"},
+				Reduce: []redis.FTAggregateReducer{reducer},
+			}},
+			SortBy: []redis.FTAggregateSortBy{{
+				FieldName: "@perDay",
+				Desc:      true,
+			}},
+		}
+
+		res, err := client.FTAggregateWithArgs(ctx, "idx1", "*", options).Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(res).ToNot(BeNil())
+		Expect(len(res.Rows)).To(BeEquivalentTo(2))
+		Expect(res.Rows[0].Fields["perDay"]).To(BeEquivalentTo("2"))
+		Expect(res.Rows[1].Fields["perDay"]).To(BeEquivalentTo("1"))
+	})
+
 	It("should FTAggregate apply", Label("search", "ftaggregate"), func() {
 		text1 := &redis.FieldSchema{FieldName: "PrimaryKey", FieldType: redis.SearchFieldTypeText, Sortable: true}
 		num1 := &redis.FieldSchema{FieldName: "CreatedDateTimeUTC", FieldType: redis.SearchFieldTypeNumeric, Sortable: true}
@@ -684,7 +783,6 @@ var _ = Describe("RediSearch commands Resp 2", Label("search"), func() {
 			Expect(res.Rows[0].Fields["age"]).To(BeEquivalentTo("19"))
 			Expect(res.Rows[1].Fields["age"]).To(BeEquivalentTo("25"))
 		}
-
 	})
 
 	It("should FTSearch SkipInitialScan", Label("search", "ftsearch"), func() {