Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Support Resp 3 Redis Search Unstable Mode #3098

Merged
merged 17 commits into from
Sep 12, 2024
Merged
Show file tree
Hide file tree
Changes from 12 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .github/wordlist.txt
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
ACLs
APIs
autoload
autoloader
autoloading
Expand Down Expand Up @@ -46,9 +47,11 @@ runtime
SHA
sharding
SETNAME
SpellCheck
SSL
struct
stunnel
SynDump
TCP
TLS
uri
Expand Down
3 changes: 3 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -182,6 +182,9 @@ rdb := redis.NewClient(&redis.Options{
})
```

#### Unstable RESP3 Structures for RediSearch Commands
When connecting with Redis using RESP3 protocol, it's important to note that some response structures aren't final yet. This is the case for more complex structures in FT.SEARCH, FT.AGGREGATE, FT.Info, FT.SpellCheck and FT.SynDump. We recommend using RESP2 when using these commands, but we plan to stabilize the RESP3-based APIs in the coming versions. You can find more guidance in the upcoming release notes.

## Contributing

Please see [out contributing guidelines](CONTRIBUTING.md) to help us improve this library!
Expand Down
17 changes: 11 additions & 6 deletions command.go
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ type Cmder interface {

readTimeout() *time.Duration
readReply(rd *proto.Reader) error

readRawReply(rd *proto.Reader) error
SetErr(error)
Err() error
}
Expand Down Expand Up @@ -122,11 +122,11 @@ func cmdString(cmd Cmder, val interface{}) string {
//------------------------------------------------------------------------------

type baseCmd struct {
ctx context.Context
args []interface{}
err error
keyPos int8

ctx context.Context
args []interface{}
err error
keyPos int8
rawVal interface{}
_readTimeout *time.Duration
}

Expand Down Expand Up @@ -197,6 +197,11 @@ func (cmd *baseCmd) setReadTimeout(d time.Duration) {
cmd._readTimeout = &d
}

func (cmd *baseCmd) readRawReply(rd *proto.Reader) (err error) {
cmd.rawVal, err = rd.ReadReply()
return err
}

//------------------------------------------------------------------------------

type Cmd struct {
Expand Down
3 changes: 3 additions & 0 deletions options.go
Original file line number Diff line number Diff line change
Expand Up @@ -153,6 +153,9 @@ type Options struct {

// Add suffix to client name. Default is empty.
IdentitySuffix string

// Enable Unstable mode for Redis Search module with RESP3.
UnstableResp3SearchModule bool
vladvildanov marked this conversation as resolved.
Show resolved Hide resolved
}

func (opt *Options) init() {
Expand Down
21 changes: 19 additions & 2 deletions redis.go
Original file line number Diff line number Diff line change
Expand Up @@ -412,6 +412,19 @@ func (c *baseClient) process(ctx context.Context, cmd Cmder) error {
return lastErr
}

func (c *baseClient) assertStableCommand(cmd Cmder) bool {
switch cmd.(type) {
case *AggregateCmd, *FTInfoCmd, *FTSpellCheckCmd, *FTSearchCmd, *FTSynDumpCmd:
if c.opt.UnstableResp3SearchModule {
return true
} else {
panic("Some RESP3 results for Redis Query Engine responses may change. Refer to the readme for guidance")
}
default:
return false
}
}

func (c *baseClient) _process(ctx context.Context, cmd Cmder, attempt int) (bool, error) {
if attempt > 0 {
if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil {
Expand All @@ -427,8 +440,12 @@ func (c *baseClient) _process(ctx context.Context, cmd Cmder, attempt int) (bool
atomic.StoreUint32(&retryTimeout, 1)
return err
}

if err := cn.WithReader(c.context(ctx), c.cmdTimeout(cmd), cmd.readReply); err != nil {
readReplyFunc := cmd.readReply
// Apply unstable RESP3 search module.
if c.opt.Protocol != 2 && c.assertStableCommand(cmd) {
vladvildanov marked this conversation as resolved.
Show resolved Hide resolved
readReplyFunc = cmd.readRawReply
}
if err := cn.WithReader(c.context(ctx), c.cmdTimeout(cmd), readReplyFunc); err != nil {
if cmd.readTimeout() == nil {
atomic.StoreUint32(&retryTimeout, 1)
} else {
Expand Down
10 changes: 6 additions & 4 deletions ring.go
Original file line number Diff line number Diff line change
Expand Up @@ -98,8 +98,9 @@ type RingOptions struct {
TLSConfig *tls.Config
Limiter Limiter

DisableIndentity bool
IdentitySuffix string
DisableIndentity bool
IdentitySuffix string
UnstableResp3SearchModule bool
}

func (opt *RingOptions) init() {
Expand Down Expand Up @@ -166,8 +167,9 @@ func (opt *RingOptions) clientOptions() *Options {
TLSConfig: opt.TLSConfig,
Limiter: opt.Limiter,

DisableIndentity: opt.DisableIndentity,
IdentitySuffix: opt.IdentitySuffix,
DisableIndentity: opt.DisableIndentity,
IdentitySuffix: opt.IdentitySuffix,
UnstableResp3SearchModule: opt.UnstableResp3SearchModule,
}
}

Expand Down
39 changes: 39 additions & 0 deletions search_commands.go
Original file line number Diff line number Diff line change
Expand Up @@ -638,6 +638,14 @@ func (cmd *AggregateCmd) Result() (*FTAggregateResult, error) {
return cmd.val, cmd.err
}

func (cmd *AggregateCmd) RawVal() interface{} {
return cmd.rawVal
}

func (cmd *AggregateCmd) RawResult() (interface{}, error) {
return cmd.rawVal, cmd.err
}

func (cmd *AggregateCmd) String() string {
return cmdString(cmd, cmd.val)
}
Expand Down Expand Up @@ -1337,6 +1345,13 @@ func (cmd *FTInfoCmd) Val() FTInfoResult {
return cmd.val
}

func (cmd *FTInfoCmd) RawVal() interface{} {
return cmd.rawVal
}

func (cmd *FTInfoCmd) RawResult() (interface{}, error) {
return cmd.rawVal, cmd.err
}
func (cmd *FTInfoCmd) readReply(rd *proto.Reader) (err error) {
n, err := rd.ReadMapLen()
if err != nil {
Expand Down Expand Up @@ -1447,6 +1462,14 @@ func (cmd *FTSpellCheckCmd) Val() []SpellCheckResult {
return cmd.val
}

func (cmd *FTSpellCheckCmd) RawVal() interface{} {
return cmd.rawVal
}

func (cmd *FTSpellCheckCmd) RawResult() (interface{}, error) {
return cmd.rawVal, cmd.err
}

func (cmd *FTSpellCheckCmd) readReply(rd *proto.Reader) (err error) {
data, err := rd.ReadSlice()
if err != nil {
Expand Down Expand Up @@ -1628,6 +1651,14 @@ func (cmd *FTSearchCmd) Val() FTSearchResult {
return cmd.val
}

func (cmd *FTSearchCmd) RawVal() interface{} {
return cmd.rawVal
}

func (cmd *FTSearchCmd) RawResult() (interface{}, error) {
return cmd.rawVal, cmd.err
}

func (cmd *FTSearchCmd) readReply(rd *proto.Reader) (err error) {
data, err := rd.ReadSlice()
if err != nil {
Expand Down Expand Up @@ -1904,6 +1935,14 @@ func (cmd *FTSynDumpCmd) Result() ([]FTSynDumpResult, error) {
return cmd.val, cmd.err
}

func (cmd *FTSynDumpCmd) RawVal() interface{} {
return cmd.rawVal
}

func (cmd *FTSynDumpCmd) RawResult() (interface{}, error) {
return cmd.rawVal, cmd.err
}

func (cmd *FTSynDumpCmd) readReply(rd *proto.Reader) error {
termSynonymPairs, err := rd.ReadSlice()
if err != nil {
Expand Down
Loading
Loading