feat(command): Add hybrid search command (#3573)

* Added hybrid search command * fixed lint, fixed some tests * lint fix * Add support for XReadGroup CLAIM argument (#3578) * Add support for XReadGroup CLAIM argument * modify tutorial tests --------- Co-authored-by: Nedyalko Dyakov <1547186+ndyakov@users.noreply.github.com> * feat(acl): add acl support and test (#3576) * feat: add acl support and command test * validate client name before kill it --------- Co-authored-by: Nedyalko Dyakov <1547186+ndyakov@users.noreply.github.com> * feat(cmd): Add support for MSetEX command (#3580) Co-authored-by: Nedyalko Dyakov <1547186+ndyakov@users.noreply.github.com> * fix(sentinel): handle empty address (#3577) * improvements * linter fixes * prevention on unnecessary allocations in case of bad configuration * Test/Benchmark, old code with safety harness preventing panic --------- Co-authored-by: manish <manish.sharma@manifestit.io> Co-authored-by: Nedyalko Dyakov <1547186+ndyakov@users.noreply.github.com> * feat: support for latency command (#3584) * support for latency command * add NonRedisEnterprise label for latency test * feat: Add support for certain slowlog commands (#3585) * Add support for certain slowlog commands * add NonRedisEnterprise label for slow reset test --------- Co-authored-by: Nedyalko Dyakov <1547186+ndyakov@users.noreply.github.com> * feat(cmd): Add CAS/CAD commands (#3583) * add cas/cad commands * feat(command): Add SetIFDEQ, SetIFDNE and *Get cmds Decided to move the *Get argument as a separate methods, since the response will be always the previous value, but in the case where the previous value is `OK` there result may be ambiguous. * fix tests * matchValue to be interface{} * Only Args approach for DelEx * use uint64 for digest, add example * test only for 8.4 * updated ft hybrid, marked as experimental * updated fthybrid and its tests * removed debugging prints * fixed lint, addressed comment * fixed issues * fixed lint * Ensure that the args are prefixed only if theres no prefix already * Removed automatic args prefixing --------- Co-authored-by: Nedyalko Dyakov <1547186+ndyakov@users.noreply.github.com> Co-authored-by: ofekshenawa <104765379+ofekshenawa@users.noreply.github.com> Co-authored-by: destinyoooo <57470814+destinyoooo@users.noreply.github.com> Co-authored-by: manish <bhardwaz007@yahoo.com> Co-authored-by: manish <manish.sharma@manifestit.io>
2025-11-14 10:22:26 +03:00 · 2025-11-11 11:48:05 +02:00
parent 6c4d77ae25
commit 9b7f1be092
2 changed files with 888 additions and 0 deletions
--- a/search_commands.go
+++ b/search_commands.go
@@ -29,6 +29,8 @@ type SearchCmdable interface {
 	FTDropIndexWithArgs(ctx context.Context, index string, options *FTDropIndexOptions) *StatusCmd
 	FTExplain(ctx context.Context, index string, query string) *StringCmd
 	FTExplainWithArgs(ctx context.Context, index string, query string, options *FTExplainOptions) *StringCmd
+	FTHybrid(ctx context.Context, index string, searchExpr string, vectorField string, vectorData Vector) *FTHybridCmd
+	FTHybridWithArgs(ctx context.Context, index string, options *FTHybridOptions) *FTHybridCmd
 	FTInfo(ctx context.Context, index string) *FTInfoCmd
 	FTSpellCheck(ctx context.Context, index string, query string) *FTSpellCheckCmd
 	FTSpellCheckWithArgs(ctx context.Context, index string, query string, options *FTSpellCheckOptions) *FTSpellCheckCmd
@@ -344,6 +346,92 @@ type FTSearchOptions struct {
 	DialectVersion int
 }

+// FTHybridCombineMethod represents the fusion method for combining search and vector results
+type FTHybridCombineMethod string
+
+const (
+	FTHybridCombineRRF      FTHybridCombineMethod = "RRF"
+	FTHybridCombineLinear   FTHybridCombineMethod = "LINEAR"
+	FTHybridCombineFunction FTHybridCombineMethod = "FUNCTION"
+)
+
+// FTHybridSearchExpression represents a search expression in hybrid search
+type FTHybridSearchExpression struct {
+	Query        string
+	Scorer       string
+	ScorerParams []interface{}
+	YieldScoreAs string
+}
+
+type FTHybridVectorMethod = string
+
+const (
+	KNN   FTHybridCombineMethod = "KNN"
+	RANGE FTHybridCombineMethod = "RANGE"
+)
+
+// FTHybridVectorExpression represents a vector expression in hybrid search
+type FTHybridVectorExpression struct {
+	VectorField  string
+	VectorData   Vector
+	Method       FTHybridVectorMethod
+	MethodParams []interface{}
+	Filter       string
+	YieldScoreAs string
+}
+
+// FTHybridCombineOptions represents options for result fusion
+type FTHybridCombineOptions struct {
+	Method       FTHybridCombineMethod
+	Count        int
+	Window       int     // For RRF
+	Constant     float64 // For RRF
+	Alpha        float64 // For LINEAR
+	Beta         float64 // For LINEAR
+	YieldScoreAs string
+}
+
+// FTHybridGroupBy represents GROUP BY functionality
+type FTHybridGroupBy struct {
+	Count        int
+	Fields       []string
+	ReduceFunc   string
+	ReduceCount  int
+	ReduceParams []interface{}
+}
+
+// FTHybridApply represents APPLY functionality
+type FTHybridApply struct {
+	Expression string
+	AsField    string
+}
+
+// FTHybridWithCursor represents cursor configuration for hybrid search
+type FTHybridWithCursor struct {
+	Count   int // Number of results to return per cursor read
+	MaxIdle int // Maximum idle time in milliseconds before cursor is automatically deleted
+}
+
+// FTHybridOptions hold options that can be passed to the FT.HYBRID command
+type FTHybridOptions struct {
+	CountExpressions  int                        // Number of search/vector expressions
+	SearchExpressions []FTHybridSearchExpression // Multiple search expressions
+	VectorExpressions []FTHybridVectorExpression // Multiple vector expressions
+	Combine           *FTHybridCombineOptions    // Fusion step options
+	Load              []string                   // Projected fields
+	GroupBy           *FTHybridGroupBy           // Aggregation grouping
+	Apply             []FTHybridApply            // Field transformations
+	SortBy            []FTSearchSortBy           // Reuse from FTSearch
+	Filter            string                     // Post-filter expression
+	LimitOffset       int                        // Result limiting
+	Limit             int
+	Params            map[string]interface{} // Parameter substitution
+	ExplainScore      bool                   // Include score explanations
+	Timeout           int                    // Runtime timeout
+	WithCursor        bool                   // Enable cursor support for large result sets
+	WithCursorOptions *FTHybridWithCursor    // Cursor configuration options
+}
+
 type FTSynDumpResult struct {
 	Term     string
 	Synonyms []string
@@ -1819,6 +1907,207 @@ func (cmd *FTSearchCmd) readReply(rd *proto.Reader) (err error) {
 	return nil
 }

+// FTHybridResult represents the result of a hybrid search operation
+type FTHybridResult struct {
+	TotalResults  int
+	Results       []map[string]interface{}
+	Warnings      []string
+	ExecutionTime float64
+}
+
+// FTHybridCursorResult represents cursor result for hybrid search
+type FTHybridCursorResult struct {
+	SearchCursorID int
+	VsimCursorID   int
+}
+
+type FTHybridCmd struct {
+	baseCmd
+	val        FTHybridResult
+	cursorVal  *FTHybridCursorResult
+	options    *FTHybridOptions
+	withCursor bool
+}
+
+func newFTHybridCmd(ctx context.Context, options *FTHybridOptions, args ...interface{}) *FTHybridCmd {
+	var withCursor bool
+	if options != nil && options.WithCursor {
+		withCursor = true
+	}
+	return &FTHybridCmd{
+		baseCmd: baseCmd{
+			ctx:  ctx,
+			args: args,
+		},
+		options:    options,
+		withCursor: withCursor,
+	}
+}
+
+func (cmd *FTHybridCmd) String() string {
+	return cmdString(cmd, cmd.val)
+}
+
+func (cmd *FTHybridCmd) SetVal(val FTHybridResult) {
+	cmd.val = val
+}
+
+func (cmd *FTHybridCmd) Result() (FTHybridResult, error) {
+	return cmd.val, cmd.err
+}
+
+func (cmd *FTHybridCmd) CursorResult() (*FTHybridCursorResult, error) {
+	return cmd.cursorVal, cmd.err
+}
+
+func (cmd *FTHybridCmd) Val() FTHybridResult {
+	return cmd.val
+}
+
+func (cmd *FTHybridCmd) CursorVal() *FTHybridCursorResult {
+	return cmd.cursorVal
+}
+
+func (cmd *FTHybridCmd) RawVal() interface{} {
+	return cmd.rawVal
+}
+
+func (cmd *FTHybridCmd) RawResult() (interface{}, error) {
+	return cmd.rawVal, cmd.err
+}
+
+func parseFTHybrid(data []interface{}, withCursor bool) (FTHybridResult, *FTHybridCursorResult, error) {
+	// Convert to map
+	resultMap := make(map[string]interface{})
+	for i := 0; i < len(data); i += 2 {
+		if i+1 < len(data) {
+			key, ok := data[i].(string)
+			if !ok {
+				return FTHybridResult{}, nil, fmt.Errorf("invalid key type at index %d", i)
+			}
+			resultMap[key] = data[i+1]
+		}
+	}
+
+	// Handle cursor result
+	if withCursor {
+		searchCursorID, ok1 := resultMap["SEARCH"].(int64)
+		vsimCursorID, ok2 := resultMap["VSIM"].(int64)
+		if !ok1 || !ok2 {
+			return FTHybridResult{}, nil, fmt.Errorf("invalid cursor result format")
+		}
+		return FTHybridResult{}, &FTHybridCursorResult{
+			SearchCursorID: int(searchCursorID),
+			VsimCursorID:   int(vsimCursorID),
+		}, nil
+	}
+
+	// Parse regular result
+	totalResults, ok := resultMap["total_results"].(int64)
+	if !ok {
+		return FTHybridResult{}, nil, fmt.Errorf("invalid total_results format")
+	}
+
+	resultsData, ok := resultMap["results"].([]interface{})
+	if !ok {
+		return FTHybridResult{}, nil, fmt.Errorf("invalid results format")
+	}
+
+	// Parse each result item
+	results := make([]map[string]interface{}, 0, len(resultsData))
+	for _, item := range resultsData {
+		// Try parsing as map[string]interface{} first (RESP3 format)
+		if itemMap, ok := item.(map[string]interface{}); ok {
+			results = append(results, itemMap)
+			continue
+		}
+
+		// Try parsing as map[interface{}]interface{} (alternative RESP3 format)
+		if rawMap, ok := item.(map[interface{}]interface{}); ok {
+			itemMap := make(map[string]interface{})
+			for k, v := range rawMap {
+				if keyStr, ok := k.(string); ok {
+					itemMap[keyStr] = v
+				}
+			}
+			results = append(results, itemMap)
+			continue
+		}
+
+		// Fall back to array format (RESP2 format - key-value pairs)
+		itemData, ok := item.([]interface{})
+		if !ok {
+			return FTHybridResult{}, nil, fmt.Errorf("invalid result item format")
+		}
+
+		itemMap := make(map[string]interface{})
+		for i := 0; i < len(itemData); i += 2 {
+			if i+1 < len(itemData) {
+				key, ok := itemData[i].(string)
+				if !ok {
+					return FTHybridResult{}, nil, fmt.Errorf("invalid item key format")
+				}
+				itemMap[key] = itemData[i+1]
+			}
+		}
+		results = append(results, itemMap)
+	}
+
+	// Parse warnings (optional field)
+	var warnings []string
+	if warningsData, ok := resultMap["warnings"].([]interface{}); ok {
+		warnings = make([]string, 0, len(warningsData))
+		for _, w := range warningsData {
+			if ws, ok := w.(string); ok {
+				warnings = append(warnings, ws)
+			}
+		}
+	}
+
+	// Parse execution time (optional field)
+	var executionTime float64
+	if execTimeVal, exists := resultMap["execution_time"]; exists {
+		switch v := execTimeVal.(type) {
+		case string:
+			var err error
+			executionTime, err = strconv.ParseFloat(v, 64)
+			if err != nil {
+				return FTHybridResult{}, nil, fmt.Errorf("invalid execution_time format: %v", err)
+			}
+		case float64:
+			executionTime = v
+		case int64:
+			executionTime = float64(v)
+		}
+	}
+
+	return FTHybridResult{
+		TotalResults:  int(totalResults),
+		Results:       results,
+		Warnings:      warnings,
+		ExecutionTime: executionTime,
+	}, nil, nil
+}
+
+func (cmd *FTHybridCmd) readReply(rd *proto.Reader) (err error) {
+	data, err := rd.ReadSlice()
+	if err != nil {
+		return err
+	}
+
+	result, cursorResult, err := parseFTHybrid(data, cmd.withCursor)
+	if err != nil {
+		return err
+	}
+
+	if cmd.withCursor {
+		cmd.cursorVal = cursorResult
+	} else {
+		cmd.val = result
+	}
+	return nil
+}
+
 // 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 about [FT.SEARCH].
@@ -2191,3 +2480,204 @@ func (c cmdable) FTTagVals(ctx context.Context, index string, field string) *Str
 	_ = c(ctx, cmd)
 	return cmd
 }
+
+// FTHybrid - Executes a hybrid search combining full-text search and vector similarity
+// The 'index' parameter specifies the index to search, 'searchExpr' is the search query,
+// 'vectorField' is the name of the vector field, and 'vectorData' is the vector to search with.
+// FTHybrid is still experimental, the command behaviour and signature may change
+func (c cmdable) FTHybrid(ctx context.Context, index string, searchExpr string, vectorField string, vectorData Vector) *FTHybridCmd {
+	options := &FTHybridOptions{
+		CountExpressions: 2,
+		SearchExpressions: []FTHybridSearchExpression{
+			{Query: searchExpr},
+		},
+		VectorExpressions: []FTHybridVectorExpression{
+			{VectorField: vectorField, VectorData: vectorData},
+		},
+	}
+	return c.FTHybridWithArgs(ctx, index, options)
+}
+
+// FTHybridWithArgs - Executes a hybrid search with advanced options
+// FTHybridWithArgs is still experimental, the command behaviour and signature may change
+func (c cmdable) FTHybridWithArgs(ctx context.Context, index string, options *FTHybridOptions) *FTHybridCmd {
+	args := []interface{}{"FT.HYBRID", index}
+
+	if options != nil {
+		// Add search expressions
+		for _, searchExpr := range options.SearchExpressions {
+			args = append(args, "SEARCH", searchExpr.Query)
+
+			if searchExpr.Scorer != "" {
+				args = append(args, "SCORER", searchExpr.Scorer)
+				if len(searchExpr.ScorerParams) > 0 {
+					args = append(args, searchExpr.ScorerParams...)
+				}
+			}
+
+			if searchExpr.YieldScoreAs != "" {
+				args = append(args, "YIELD_SCORE_AS", searchExpr.YieldScoreAs)
+			}
+		}
+
+		// Add vector expressions
+		for _, vectorExpr := range options.VectorExpressions {
+			args = append(args, "VSIM", "@"+vectorExpr.VectorField)
+
+			// For FT.HYBRID, we need to send just the raw vector bytes, not the Value() format
+			// Value() returns [format, data] but FT.HYBRID expects just the blob
+			vectorValue := vectorExpr.VectorData.Value()
+			if len(vectorValue) >= 2 {
+				// vectorValue is [format, data, ...] - we only want the data part
+				args = append(args, vectorValue[1])
+			} else {
+				// Fallback for unexpected format
+				args = append(args, vectorValue...)
+			}
+
+			if vectorExpr.Method != "" {
+				args = append(args, vectorExpr.Method)
+				if len(vectorExpr.MethodParams) > 0 {
+					// MethodParams should be key-value pairs, count them
+					args = append(args, len(vectorExpr.MethodParams))
+					args = append(args, vectorExpr.MethodParams...)
+				}
+			}
+
+			if vectorExpr.Filter != "" {
+				args = append(args, "FILTER", vectorExpr.Filter)
+			}
+
+			if vectorExpr.YieldScoreAs != "" {
+				args = append(args, "YIELD_SCORE_AS", vectorExpr.YieldScoreAs)
+			}
+		}
+
+		// Add combine/fusion options
+		if options.Combine != nil {
+			// Build combine parameters
+			combineParams := []interface{}{}
+
+			switch options.Combine.Method {
+			case FTHybridCombineRRF:
+				if options.Combine.Window > 0 {
+					combineParams = append(combineParams, "WINDOW", options.Combine.Window)
+				}
+				if options.Combine.Constant > 0 {
+					combineParams = append(combineParams, "CONSTANT", options.Combine.Constant)
+				}
+			case FTHybridCombineLinear:
+				if options.Combine.Alpha > 0 {
+					combineParams = append(combineParams, "ALPHA", options.Combine.Alpha)
+				}
+				if options.Combine.Beta > 0 {
+					combineParams = append(combineParams, "BETA", options.Combine.Beta)
+				}
+			}
+
+			if options.Combine.YieldScoreAs != "" {
+				combineParams = append(combineParams, "YIELD_SCORE_AS", options.Combine.YieldScoreAs)
+			}
+
+			// Add COMBINE with method and parameter count
+			args = append(args, "COMBINE", string(options.Combine.Method))
+			if len(combineParams) > 0 {
+				args = append(args, len(combineParams))
+				args = append(args, combineParams...)
+			}
+		}
+
+		// Add LOAD (projected fields)
+		if len(options.Load) > 0 {
+			args = append(args, "LOAD", len(options.Load))
+			for _, field := range options.Load {
+				args = append(args, field)
+			}
+		}
+
+		// Add GROUPBY
+		if options.GroupBy != nil {
+			args = append(args, "GROUPBY", options.GroupBy.Count)
+			for _, field := range options.GroupBy.Fields {
+				args = append(args, field)
+			}
+			if options.GroupBy.ReduceFunc != "" {
+				args = append(args, "REDUCE", options.GroupBy.ReduceFunc, options.GroupBy.ReduceCount)
+				args = append(args, options.GroupBy.ReduceParams...)
+			}
+		}
+
+		// Add APPLY transformations
+		for _, apply := range options.Apply {
+			args = append(args, "APPLY", apply.Expression, "AS", apply.AsField)
+		}
+
+		// Add SORTBY
+		if len(options.SortBy) > 0 {
+			sortByOptions := []interface{}{}
+			for _, sortBy := range options.SortBy {
+				sortByOptions = append(sortByOptions, sortBy.FieldName)
+				if sortBy.Asc && sortBy.Desc {
+					cmd := newFTHybridCmd(ctx, options, args...)
+					cmd.SetErr(fmt.Errorf("FT.HYBRID: 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, "SORTBY", len(sortByOptions))
+			args = append(args, sortByOptions...)
+		}
+
+		// Add FILTER (post-filter)
+		if options.Filter != "" {
+			args = append(args, "FILTER", options.Filter)
+		}
+
+		// Add LIMIT
+		if options.LimitOffset >= 0 && options.Limit > 0 || options.LimitOffset > 0 && options.Limit == 0 {
+			args = append(args, "LIMIT", options.LimitOffset, options.Limit)
+		}
+
+		// Add PARAMS
+		if len(options.Params) > 0 {
+			args = append(args, "PARAMS", len(options.Params)*2)
+			for key, value := range options.Params {
+				// Parameter keys should already have '$' prefix from the user
+				// Don't add it again if it's already there
+				args = append(args, key, value)
+			}
+		}
+
+		// Add EXPLAINSCORE
+		if options.ExplainScore {
+			args = append(args, "EXPLAINSCORE")
+		}
+
+		// Add TIMEOUT
+		if options.Timeout > 0 {
+			args = append(args, "TIMEOUT", options.Timeout)
+		}
+
+		// Add WITHCURSOR support
+		if options.WithCursor {
+			args = append(args, "WITHCURSOR")
+			if options.WithCursorOptions != nil {
+				if options.WithCursorOptions.Count > 0 {
+					args = append(args, "COUNT", options.WithCursorOptions.Count)
+				}
+				if options.WithCursorOptions.MaxIdle > 0 {
+					args = append(args, "MAXIDLE", options.WithCursorOptions.MaxIdle)
+				}
+			}
+		}
+	}
+
+	cmd := newFTHybridCmd(ctx, options, args...)
+	_ = c(ctx, cmd)
+	return cmd
+}