Merge a5fdfa2ec8 into e63669e170

Bumps [rojopolis/spellcheck-github-actions](https://github.com/rojopolis/spellcheck-github-actions) from 0.40.0 to 0.45.0.
- [Release notes](https://github.com/rojopolis/spellcheck-github-actions/releases)
- [Changelog](https://github.com/rojopolis/spellcheck-github-actions/blob/master/CHANGELOG.md)
- [Commits](https://github.com/rojopolis/spellcheck-github-actions/compare/0.40.0...0.45.0)

---
updated-dependencies:
- dependency-name: rojopolis/spellcheck-github-actions
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: ofekshenawa <104765379+ofekshenawa@users.noreply.github.com>

.github/wordlist.txt

@@ -54,6 +54,7 @@ stunnel
 SynDump
 TCP
 TLS
+UnstableResp
 uri
 URI
 url
@@ -62,3 +63,5 @@ RedisStack
 RedisGears
 RedisTimeseries
 RediSearch
+RawResult
+RawVal

.github/workflows/spellcheck.yml

@@ -8,7 +8,7 @@ jobs:
       - name: Checkout
         uses: actions/checkout@v4
       - name: Check Spelling
-        uses: rojopolis/spellcheck-github-actions@0.40.0
+        uses: rojopolis/spellcheck-github-actions@0.45.0
         with:
           config_path: .github/spellcheck-settings.yml
           task_name: Markdown

README.md

@@ -186,6 +186,21 @@ rdb := redis.NewClient(&redis.Options{
 #### Unstable RESP3 Structures for RediSearch Commands
 When integrating Redis with application functionalities using RESP3, it's important to note that some response structures aren't final yet. This is especially true for more complex structures like search and query results. We recommend using RESP2 when using the search and query capabilities, but we plan to stabilize the RESP3-based API-s in the coming versions. You can find more guidance in the upcoming release notes.
 
+To enable unstable RESP3, set the option in your client configuration:
+
+```go
+redis.NewClient(&redis.Options{
+			UnstableResp3: true,
+		})
+```
+**Note:** When UnstableResp3 mode is enabled, it's necessary to use RawResult() and RawVal() to retrieve a raw data.
+          Since, raw response is the only option for unstable search commands Val() and Result() calls wouldn't have any affect on them:
+
+```go
+res1, err := client.FTSearchWithArgs(ctx, "txt", "foo bar", &redis.FTSearchOptions{}).RawResult()
+val1 := client.FTSearchWithArgs(ctx, "txt", "foo bar", &redis.FTSearchOptions{}).RawVal()
+```
+
 ## Contributing
 
 Please see [out contributing guidelines](CONTRIBUTING.md) to help us improve this library!

command.go

@@ -30,6 +30,9 @@ type Cmder interface {
 	// e.g. "set k v ex 10" -> "[set k v ex 10]".
 	Args() []interface{}
 
+	//all keys in command.
+	Keys() []string
+
 	// format request and response string.
 	// e.g. "set k v ex 10" -> "set k v ex 10: OK", "get k" -> "get k: v".
 	String() string
@@ -126,6 +129,7 @@ type baseCmd struct {
 	args         []interface{}
 	err          error
 	keyPos       int8
+	keys         []string
 	rawVal       interface{}
 	_readTimeout *time.Duration
 }
@@ -159,6 +163,10 @@ func (cmd *baseCmd) Args() []interface{} {
 	return cmd.args
 }
 
+func (cmd *baseCmd) Keys() []string {
+	return cmd.keys
+}
+
 func (cmd *baseCmd) stringArg(pos int) string {
 	if pos < 0 || pos >= len(cmd.args) {
 		return ""
@@ -167,6 +175,8 @@ func (cmd *baseCmd) stringArg(pos int) string {
 	switch v := arg.(type) {
 	case string:
 		return v
+	case []byte:
+		return string(v)
 	default:
 		// TODO: consider using appendArg
 		return fmt.Sprint(v)
@@ -202,6 +212,22 @@ func (cmd *baseCmd) readRawReply(rd *proto.Reader) (err error) {
 	return err
 }
 
+// getInterleavedArguments returns arguments at even indices starting from index 1.
+func (cmd *baseCmd) getInterleavedArguments() []string {
+	return cmd.getInterleavedArgumentsWithOffset(1)
+}
+
+// getInterleavedArgumentsWithOffset returns arguments at even indices starting from the specified offset.
+func (cmd *baseCmd) getInterleavedArgumentsWithOffset(offset int) []string {
+	var matchingArguments []string
+	for i := offset; i < len(cmd.args); i += 2 {
+		if arg, ok := cmd.args[i].(string); ok {
+			matchingArguments = append(matchingArguments, arg)
+		}
+	}
+	return matchingArguments
+}
+
 //------------------------------------------------------------------------------
 
 type Cmd struct {
@@ -1403,11 +1429,18 @@ func (cmd *MapStringSliceInterfaceCmd) Val() map[string][]interface{} {
 }
 
 func (cmd *MapStringSliceInterfaceCmd) readReply(rd *proto.Reader) (err error) {
+	readType, err := rd.PeekReplyType()
+	if err != nil {
+		return err
+	}
+
+	cmd.val = make(map[string][]interface{})
+
+	if readType == proto.RespMap {
 		n, err := rd.ReadMapLen()
 		if err != nil {
 			return err
 		}
-	cmd.val = make(map[string][]interface{}, n)
 		for i := 0; i < n; i++ {
 			k, err := rd.ReadString()
 			if err != nil {
@@ -1426,6 +1459,35 @@ func (cmd *MapStringSliceInterfaceCmd) readReply(rd *proto.Reader) (err error) {
 				cmd.val[k][j] = value
 			}
 		}
+	} else if readType == proto.RespArray {
+		// RESP2 response
+		n, err := rd.ReadArrayLen()
+		if err != nil {
+			return err
+		}
+
+		for i := 0; i < n; i++ {
+			// Each entry in this array is itself an array with key details
+			itemLen, err := rd.ReadArrayLen()
+			if err != nil {
+				return err
+			}
+
+			key, err := rd.ReadString()
+			if err != nil {
+				return err
+			}
+			cmd.val[key] = make([]interface{}, 0, itemLen-1)
+			for j := 1; j < itemLen; j++ {
+				// Read the inner array for timestamp-value pairs
+				data, err := rd.ReadReply()
+				if err != nil {
+					return err
+				}
+				cmd.val[key] = append(cmd.val[key], data)
+			}
+		}
+	}
 
 	return nil
 }

commands_test.go

@@ -7279,6 +7279,67 @@ var _ = Describe("Commands", func() {
 	})
 })
 
+var _ = Describe("Keys Extraction Tests", func() {
+	var client *redis.Client
+	var ctx = context.TODO()
+
+	BeforeEach(func() {
+		client = redis.NewClient(redisOptions())
+		Expect(client.FlushDB(ctx).Err()).NotTo(HaveOccurred())
+	})
+
+	AfterEach(func() {
+		Expect(client.Close()).NotTo(HaveOccurred())
+	})
+
+	// STRING COMMANDS
+
+	It("should test Append command", func() {
+		cmd := client.Append(ctx, "key1", "value")
+		Expect(cmd.Keys()).To(Equal([]string{"key1"}))
+	})
+
+	It("should test Decr command", func() {
+		cmd := client.Decr(ctx, "key1")
+		Expect(cmd.Keys()).To(Equal([]string{"key1"}))
+	})
+
+	It("should test Get command", func() {
+		cmd := client.Get(ctx, "key1")
+		Expect(cmd.Keys()).To(Equal([]string{"key1"}))
+	})
+
+	It("should test MGet command", func() {
+		cmd := client.MGet(ctx, "key1", "key2", "key3")
+		Expect(cmd.Keys()).To(Equal([]string{"key1", "key2", "key3"}))
+	})
+
+	It("should test Set command", func() {
+		cmd := client.Set(ctx, "key1", "value", time.Second)
+		Expect(cmd.Keys()).To(Equal([]string{"key1"}))
+	})
+
+	It("should test MSet command", func() {
+		cmd := client.MSet(ctx, "key1", "value1", "key2", "value2")
+		Expect(cmd.Keys()).To(Equal([]string{"key1", "key2"}))
+	})
+
+	It("should test IncrBy command", func() {
+		cmd := client.IncrBy(ctx, "key1", 10)
+		Expect(cmd.Keys()).To(Equal([]string{"key1"}))
+	})
+
+	It("should test SetNX command", func() {
+		cmd := client.SetNX(ctx, "key1", "value", time.Second)
+		Expect(cmd.Keys()).To(Equal([]string{"key1"}))
+	})
+
+	It("should test GetDel command", func() {
+		cmd := client.GetDel(ctx, "key1")
+		Expect(cmd.Keys()).To(Equal([]string{"key1"}))
+	})
+})
+
 type numberStruct struct {
 	Number int
 }

doctests/home_json_example_test.go

@@ -0,0 +1,199 @@
+// EXAMPLE: go_home_json
+// HIDE_START
+package example_commands_test
+
+// HIDE_END
+// STEP_START import
+import (
+	"context"
+	"fmt"
+	"sort"
+
+	"github.com/redis/go-redis/v9"
+)
+
+// STEP_END
+
+func ExampleClient_search_json() {
+	// STEP_START connect
+	ctx := context.Background()
+
+	rdb := redis.NewClient(&redis.Options{
+		Addr:     "localhost:6379",
+		Password: "", // no password docs
+		DB:       0,  // use default DB
+		Protocol: 2,
+	})
+	// STEP_END
+	// REMOVE_START
+	rdb.Del(ctx, "user:1", "user:2", "user:3")
+	rdb.FTDropIndex(ctx, "idx:users")
+	// REMOVE_END
+
+	// STEP_START create_data
+	user1 := map[string]interface{}{
+		"name":  "Paul John",
+		"email": "paul.john@example.com",
+		"age":   42,
+		"city":  "London",
+	}
+
+	user2 := map[string]interface{}{
+		"name":  "Eden Zamir",
+		"email": "eden.zamir@example.com",
+		"age":   29,
+		"city":  "Tel Aviv",
+	}
+
+	user3 := map[string]interface{}{
+		"name":  "Paul Zamir",
+		"email": "paul.zamir@example.com",
+		"age":   35,
+		"city":  "Tel Aviv",
+	}
+	// STEP_END
+
+	// STEP_START make_index
+	_, err := rdb.FTCreate(
+		ctx,
+		"idx:users",
+		// Options:
+		&redis.FTCreateOptions{
+			OnJSON: true,
+			Prefix: []interface{}{"user:"},
+		},
+		// Index schema fields:
+		&redis.FieldSchema{
+			FieldName: "$.name",
+			As:        "name",
+			FieldType: redis.SearchFieldTypeText,
+		},
+		&redis.FieldSchema{
+			FieldName: "$.city",
+			As:        "city",
+			FieldType: redis.SearchFieldTypeTag,
+		},
+		&redis.FieldSchema{
+			FieldName: "$.age",
+			As:        "age",
+			FieldType: redis.SearchFieldTypeNumeric,
+		},
+	).Result()
+
+	if err != nil {
+		panic(err)
+	}
+	// STEP_END
+
+	// STEP_START add_data
+	_, err = rdb.JSONSet(ctx, "user:1", "$", user1).Result()
+
+	if err != nil {
+		panic(err)
+	}
+
+	_, err = rdb.JSONSet(ctx, "user:2", "$", user2).Result()
+
+	if err != nil {
+		panic(err)
+	}
+
+	_, err = rdb.JSONSet(ctx, "user:3", "$", user3).Result()
+
+	if err != nil {
+		panic(err)
+	}
+	// STEP_END
+
+	// STEP_START query1
+	findPaulResult, err := rdb.FTSearch(
+		ctx,
+		"idx:users",
+		"Paul @age:[30 40]",
+	).Result()
+
+	if err != nil {
+		panic(err)
+	}
+
+	fmt.Println(findPaulResult)
+	// >>> {1 [{user:3 <nil> <nil> <nil> map[$:{"age":35,"city":"Tel Aviv"...
+	// STEP_END
+
+	// STEP_START query2
+	citiesResult, err := rdb.FTSearchWithArgs(
+		ctx,
+		"idx:users",
+		"Paul",
+		&redis.FTSearchOptions{
+			Return: []redis.FTSearchReturn{
+				{
+					FieldName: "$.city",
+					As:        "city",
+				},
+			},
+		},
+	).Result()
+
+	if err != nil {
+		panic(err)
+	}
+
+	sort.Slice(citiesResult.Docs, func(i, j int) bool {
+		return citiesResult.Docs[i].Fields["city"] < citiesResult.Docs[j].Fields["city"]
+	})
+
+	for _, result := range citiesResult.Docs {
+		fmt.Println(result.Fields["city"])
+	}
+	// >>> London
+	// >>> Tel Aviv
+	// STEP_END
+
+	// STEP_START query3
+	aggOptions := redis.FTAggregateOptions{
+		GroupBy: []redis.FTAggregateGroupBy{
+			{
+				Fields: []interface{}{"@city"},
+				Reduce: []redis.FTAggregateReducer{
+					{
+						Reducer: redis.SearchCount,
+						As:      "count",
+					},
+				},
+			},
+		},
+	}
+
+	aggResult, err := rdb.FTAggregateWithArgs(
+		ctx,
+		"idx:users",
+		"*",
+		&aggOptions,
+	).Result()
+
+	if err != nil {
+		panic(err)
+	}
+
+	sort.Slice(aggResult.Rows, func(i, j int) bool {
+		return aggResult.Rows[i].Fields["city"].(string) <
+			aggResult.Rows[j].Fields["city"].(string)
+	})
+
+	for _, row := range aggResult.Rows {
+		fmt.Printf("%v - %v\n",
+			row.Fields["city"], row.Fields["count"],
+		)
+	}
+	// >>> City: London - 1
+	// >>> City: Tel Aviv - 2
+	// STEP_END
+
+	// Output:
+	// {1 [{user:3 <nil> <nil> <nil> map[$:{"age":35,"city":"Tel Aviv","email":"paul.zamir@example.com","name":"Paul Zamir"}]}]}
+	// London
+	// Tel Aviv
+	// London - 1
+	// Tel Aviv - 2
+}

osscluster.go

@@ -21,6 +21,10 @@ import (
 	"github.com/redis/go-redis/v9/internal/rand"
 )
 
+const (
+	minLatencyMeasurementInterval = 10 * time.Second
+)
+
 var errClusterNoNodes = fmt.Errorf("redis: cluster has no nodes")
 
 // ClusterOptions are used to configure a cluster client and should be
@@ -316,6 +320,10 @@ type clusterNode struct {
 	latency    uint32 // atomic
 	generation uint32 // atomic
 	failing    uint32 // atomic
+
+	// last time the latency measurement was performed for the node, stored in nanoseconds
+	// from epoch
+	lastLatencyMeasurement int64 // atomic
 }
 
 func newClusterNode(clOpt *ClusterOptions, addr string) *clusterNode {
@@ -368,6 +376,7 @@ func (n *clusterNode) updateLatency() {
 		latency = float64(dur) / float64(successes)
 	}
 	atomic.StoreUint32(&n.latency, uint32(latency+0.5))
+	n.SetLastLatencyMeasurement(time.Now())
 }
 
 func (n *clusterNode) Latency() time.Duration {
@@ -397,6 +406,10 @@ func (n *clusterNode) Generation() uint32 {
 	return atomic.LoadUint32(&n.generation)
 }
 
+func (n *clusterNode) LastLatencyMeasurement() int64 {
+	return atomic.LoadInt64(&n.lastLatencyMeasurement)
+}
+
 func (n *clusterNode) SetGeneration(gen uint32) {
 	for {
 		v := atomic.LoadUint32(&n.generation)
@@ -406,6 +419,15 @@ func (n *clusterNode) SetGeneration(gen uint32) {
 	}
 }
 
+func (n *clusterNode) SetLastLatencyMeasurement(t time.Time) {
+	for {
+		v := atomic.LoadInt64(&n.lastLatencyMeasurement)
+		if t.UnixNano() < v || atomic.CompareAndSwapInt64(&n.lastLatencyMeasurement, v, t.UnixNano()) {
+			break
+		}
+	}
+}
+
 //------------------------------------------------------------------------------
 
 type clusterNodes struct {
@@ -493,10 +515,11 @@ func (c *clusterNodes) GC(generation uint32) {
 	c.mu.Lock()
 
 	c.activeAddrs = c.activeAddrs[:0]
+	now := time.Now()
 	for addr, node := range c.nodes {
 		if node.Generation() >= generation {
 			c.activeAddrs = append(c.activeAddrs, addr)
-			if c.opt.RouteByLatency {
+			if c.opt.RouteByLatency && node.LastLatencyMeasurement() < now.Add(-minLatencyMeasurementInterval).UnixNano() {
 				go node.updateLatency()
 			}
 			continue

osscluster_test.go

@@ -653,6 +653,32 @@ var _ = Describe("ClusterClient", func() {
 			Expect(client.Close()).NotTo(HaveOccurred())
 		})
 
+		It("determines hash slots correctly for generic commands", func() {
+			opt := redisClusterOptions()
+			opt.MaxRedirects = -1
+			client := cluster.newClusterClient(ctx, opt)
+
+			err := client.Do(ctx, "GET", "A").Err()
+			Expect(err).To(Equal(redis.Nil))
+
+			err = client.Do(ctx, []byte("GET"), []byte("A")).Err()
+			Expect(err).To(Equal(redis.Nil))
+
+			Eventually(func() error {
+				return client.SwapNodes(ctx, "A")
+			}, 30*time.Second).ShouldNot(HaveOccurred())
+
+			err = client.Do(ctx, "GET", "A").Err()
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("MOVED"))
+
+			err = client.Do(ctx, []byte("GET"), []byte("A")).Err()
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("MOVED"))
+
+			Expect(client.Close()).NotTo(HaveOccurred())
+		})
+
 		It("follows node redirection immediately", func() {
 			// Configure retry backoffs far in excess of the expected duration of redirection
 			opt := redisClusterOptions()

probabilistic.go

@@ -319,38 +319,70 @@ func (cmd *BFInfoCmd) Result() (BFInfo, error) {
 }
 
 func (cmd *BFInfoCmd) readReply(rd *proto.Reader) (err error) {
+	result := BFInfo{}
+
+	// Create a mapping from key names to pointers of struct fields
+	respMapping := map[string]*int64{
+		"Capacity":                 &result.Capacity,
+		"CAPACITY":                 &result.Capacity,
+		"Size":                     &result.Size,
+		"SIZE":                     &result.Size,
+		"Number of filters":        &result.Filters,
+		"FILTERS":                  &result.Filters,
+		"Number of items inserted": &result.ItemsInserted,
+		"ITEMS":                    &result.ItemsInserted,
+		"Expansion rate":           &result.ExpansionRate,
+		"EXPANSION":                &result.ExpansionRate,
+	}
+
+	// Helper function to read and assign a value based on the key
+	readAndAssignValue := func(key string) error {
+		fieldPtr, exists := respMapping[key]
+		if !exists {
+			return fmt.Errorf("redis: BLOOM.INFO unexpected key %s", key)
+		}
+
+		// Read the integer and assign to the field via pointer dereferencing
+		val, err := rd.ReadInt()
+		if err != nil {
+			return err
+		}
+		*fieldPtr = val
+		return nil
+	}
+
+	readType, err := rd.PeekReplyType()
+	if err != nil {
+		return err
+	}
+
+	if len(cmd.args) > 2 && readType == proto.RespArray {
+		n, err := rd.ReadArrayLen()
+		if err != nil {
+			return err
+		}
+		if key, ok := cmd.args[2].(string); ok && n == 1 {
+			if err := readAndAssignValue(key); err != nil {
+				return err
+			}
+		} else {
+			return fmt.Errorf("redis: BLOOM.INFO invalid argument key type")
+		}
+	} else {
 		n, err := rd.ReadMapLen()
 		if err != nil {
 			return err
 		}
-
-	var key string
-	var result BFInfo
-	for f := 0; f < n; f++ {
-		key, err = rd.ReadString()
+		for i := 0; i < n; i++ {
+			key, err := rd.ReadString()
 			if err != nil {
 				return err
 			}
-
-		switch key {
-		case "Capacity":
-			result.Capacity, err = rd.ReadInt()
-		case "Size":
-			result.Size, err = rd.ReadInt()
-		case "Number of filters":
-			result.Filters, err = rd.ReadInt()
-		case "Number of items inserted":
-			result.ItemsInserted, err = rd.ReadInt()
-		case "Expansion rate":
-			result.ExpansionRate, err = rd.ReadInt()
-		default:
-			return fmt.Errorf("redis: BLOOM.INFO unexpected key %s", key)
-		}
-
-		if err != nil {
+			if err := readAndAssignValue(key); err != nil {
 				return err
 			}
 		}
+	}
 
 	cmd.val = result
 	return nil

probabilistic_test.go

@@ -13,15 +13,32 @@ import (
 
 var _ = Describe("Probabilistic commands", Label("probabilistic"), func() {
 	ctx := context.TODO()
+
+	setupRedisClient := func(protocolVersion int) *redis.Client {
+		return redis.NewClient(&redis.Options{
+			Addr:     "localhost:6379",
+			DB:       0,
+			Protocol: protocolVersion,
+		})
+	}
+
+	protocols := []int{2, 3}
+	for _, protocol := range protocols {
+		protocol := protocol // capture loop variable for each context
+
+		Context(fmt.Sprintf("with protocol version %d", protocol), func() {
 			var client *redis.Client
 
 			BeforeEach(func() {
-		client = redis.NewClient(&redis.Options{Addr: ":6379"})
-		Expect(client.FlushDB(ctx).Err()).NotTo(HaveOccurred())
+				client = setupRedisClient(protocol)
+				Expect(client.FlushAll(ctx).Err()).NotTo(HaveOccurred())
 			})
 
 			AfterEach(func() {
-		Expect(client.Close()).NotTo(HaveOccurred())
+				if client != nil {
+					client.FlushDB(ctx)
+					client.Close()
+				}
 			})
 
 			Describe("bloom", Label("bloom"), func() {
@@ -117,7 +134,7 @@ var _ = Describe("Probabilistic commands", Label("probabilistic"), func() {
 						NoCreate:   true,
 					}
 
-			resultInsert, err := client.BFInsert(ctx, "testbf1", options, "item1").Result()
+					_, err := client.BFInsert(ctx, "testbf1", options, "item1").Result()
 					Expect(err).To(HaveOccurred())
 					Expect(err).To(MatchError("ERR not found"))
 
@@ -129,7 +146,7 @@ var _ = Describe("Probabilistic commands", Label("probabilistic"), func() {
 						NoCreate:   false,
 					}
 
-			resultInsert, err = client.BFInsert(ctx, "testbf1", options, "item1").Result()
+					resultInsert, err := client.BFInsert(ctx, "testbf1", options, "item1").Result()
 					Expect(err).NotTo(HaveOccurred())
 					Expect(len(resultInsert)).To(BeEquivalentTo(1))
 
@@ -396,7 +413,7 @@ var _ = Describe("Probabilistic commands", Label("probabilistic"), func() {
 						NoCreate: true,
 					}
 
-			result, err := client.CFInsertNX(ctx, "testcf1", args, "item1", "item2", "item2").Result()
+					_, err := client.CFInsertNX(ctx, "testcf1", args, "item1", "item2", "item2").Result()
 					Expect(err).To(HaveOccurred())
 
 					args = &redis.CFInsertOptions{
@@ -404,7 +421,7 @@ var _ = Describe("Probabilistic commands", Label("probabilistic"), func() {
 						NoCreate: false,
 					}
 
-			result, err = client.CFInsertNX(ctx, "testcf2", args, "item1", "item2", "item2").Result()
+					result, err := client.CFInsertNX(ctx, "testcf2", args, "item1", "item2", "item2").Result()
 					Expect(err).NotTo(HaveOccurred())
 					Expect(len(result)).To(BeEquivalentTo(3))
 					Expect(result[0]).To(BeEquivalentTo(int64(1)))
@@ -730,4 +747,6 @@ var _ = Describe("Probabilistic commands", Label("probabilistic"), func() {
 					Expect(max).To(BeEquivalentTo(float64(140)))
 				})
 			})
+		})
+	}
 })

redis.go

@@ -176,8 +176,6 @@ func (hs *hooksMixin) withProcessPipelineHook(
 }
 
 func (hs *hooksMixin) dialHook(ctx context.Context, network, addr string) (net.Conn, error) {
-	hs.hooksMu.Lock()
-	defer hs.hooksMu.Unlock()
 	return hs.current.dial(ctx, network, addr)
 }
 

redis_test.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"sync"
 	"testing"
 	"time"
 
@@ -633,3 +634,67 @@ var _ = Describe("Hook with MinIdleConns", func() {
 		}))
 	})
 })
+
+var _ = Describe("Dialer connection timeouts", func() {
+	var client *redis.Client
+
+	const dialSimulatedDelay = 1 * time.Second
+
+	BeforeEach(func() {
+		options := redisOptions()
+		options.Dialer = func(ctx context.Context, network, addr string) (net.Conn, error) {
+			// Simulated slow dialer.
+			// Note that the following sleep is deliberately not context-aware.
+			time.Sleep(dialSimulatedDelay)
+			return net.Dial("tcp", options.Addr)
+		}
+		options.MinIdleConns = 1
+		client = redis.NewClient(options)
+	})
+
+	AfterEach(func() {
+		err := client.Close()
+		Expect(err).NotTo(HaveOccurred())
+	})
+
+	It("does not contend on connection dial for concurrent commands", func() {
+		var wg sync.WaitGroup
+
+		const concurrency = 10
+
+		durations := make(chan time.Duration, concurrency)
+		errs := make(chan error, concurrency)
+
+		start := time.Now()
+		wg.Add(concurrency)
+
+		for i := 0; i < concurrency; i++ {
+			go func() {
+				defer wg.Done()
+
+				start := time.Now()
+				err := client.Ping(ctx).Err()
+				durations <- time.Since(start)
+				errs <- err
+			}()
+		}
+
+		wg.Wait()
+		close(durations)
+		close(errs)
+
+		// All commands should eventually succeed, after acquiring a connection.
+		for err := range errs {
+			Expect(err).NotTo(HaveOccurred())
+		}
+
+		// Each individual command should complete within the simulated dial duration bound.
+		for duration := range durations {
+			Expect(duration).To(BeNumerically("<", 2*dialSimulatedDelay))
+		}
+
+		// Due to concurrent execution, the entire test suite should also complete within
+		// the same dial duration bound applied for individual commands.
+		Expect(time.Since(start)).To(BeNumerically("<", 2*dialSimulatedDelay))
+	})
+})

string_commands.go

@@ -32,18 +32,21 @@ type StringCmdable interface {
 
 func (c cmdable) Append(ctx context.Context, key, value string) *IntCmd {
 	cmd := NewIntCmd(ctx, "append", key, value)
+	cmd.keys = append(cmd.keys, key)
 	_ = c(ctx, cmd)
 	return cmd
 }
 
 func (c cmdable) Decr(ctx context.Context, key string) *IntCmd {
 	cmd := NewIntCmd(ctx, "decr", key)
+	cmd.keys = append(cmd.keys, key)
 	_ = c(ctx, cmd)
 	return cmd
 }
 
 func (c cmdable) DecrBy(ctx context.Context, key string, decrement int64) *IntCmd {
 	cmd := NewIntCmd(ctx, "decrby", key, decrement)
+	cmd.keys = append(cmd.keys, key)
 	_ = c(ctx, cmd)
 	return cmd
 }
@@ -51,18 +54,21 @@ func (c cmdable) DecrBy(ctx context.Context, key string, decrement int64) *IntCm
 // Get Redis `GET key` command. It returns redis.Nil error when key does not exist.
 func (c cmdable) Get(ctx context.Context, key string) *StringCmd {
 	cmd := NewStringCmd(ctx, "get", key)
+	cmd.keys = append(cmd.keys, key)
 	_ = c(ctx, cmd)
 	return cmd
 }
 
 func (c cmdable) GetRange(ctx context.Context, key string, start, end int64) *StringCmd {
 	cmd := NewStringCmd(ctx, "getrange", key, start, end)
+	cmd.keys = append(cmd.keys, key)
 	_ = c(ctx, cmd)
 	return cmd
 }
 
 func (c cmdable) GetSet(ctx context.Context, key string, value interface{}) *StringCmd {
 	cmd := NewStringCmd(ctx, "getset", key, value)
+	cmd.keys = append(cmd.keys, key)
 	_ = c(ctx, cmd)
 	return cmd
 }
@@ -83,6 +89,7 @@ func (c cmdable) GetEx(ctx context.Context, key string, expiration time.Duration
 	}
 
 	cmd := NewStringCmd(ctx, args...)
+	cmd.keys = append(cmd.keys, key)
 	_ = c(ctx, cmd)
 	return cmd
 }
@@ -90,24 +97,28 @@ func (c cmdable) GetEx(ctx context.Context, key string, expiration time.Duration
 // GetDel redis-server version >= 6.2.0.
 func (c cmdable) GetDel(ctx context.Context, key string) *StringCmd {
 	cmd := NewStringCmd(ctx, "getdel", key)
+	cmd.keys = append(cmd.keys, key)
 	_ = c(ctx, cmd)
 	return cmd
 }
 
 func (c cmdable) Incr(ctx context.Context, key string) *IntCmd {
 	cmd := NewIntCmd(ctx, "incr", key)
+	cmd.keys = append(cmd.keys, key)
 	_ = c(ctx, cmd)
 	return cmd
 }
 
 func (c cmdable) IncrBy(ctx context.Context, key string, value int64) *IntCmd {
 	cmd := NewIntCmd(ctx, "incrby", key, value)
+	cmd.keys = append(cmd.keys, key)
 	_ = c(ctx, cmd)
 	return cmd
 }
 
 func (c cmdable) IncrByFloat(ctx context.Context, key string, value float64) *FloatCmd {
 	cmd := NewFloatCmd(ctx, "incrbyfloat", key, value)
+	cmd.keys = append(cmd.keys, key)
 	_ = c(ctx, cmd)
 	return cmd
 }
@@ -125,6 +136,7 @@ func (c cmdable) MGet(ctx context.Context, keys ...string) *SliceCmd {
 		args[1+i] = key
 	}
 	cmd := NewSliceCmd(ctx, args...)
+	cmd.keys = append(cmd.keys, keys...)
 	_ = c(ctx, cmd)
 	return cmd
 }
@@ -139,6 +151,7 @@ func (c cmdable) MSet(ctx context.Context, values ...interface{}) *StatusCmd {
 	args[0] = "mset"
 	args = appendArgs(args, values)
 	cmd := NewStatusCmd(ctx, args...)
+	cmd.keys = append(cmd.keys, cmd.getInterleavedArguments()...)
 	_ = c(ctx, cmd)
 	return cmd
 }
@@ -153,6 +166,7 @@ func (c cmdable) MSetNX(ctx context.Context, values ...interface{}) *BoolCmd {
 	args[0] = "msetnx"
 	args = appendArgs(args, values)
 	cmd := NewBoolCmd(ctx, args...)
+	cmd.keys = append(cmd.keys, cmd.getInterleavedArguments()...)
 	_ = c(ctx, cmd)
 	return cmd
 }
@@ -179,6 +193,7 @@ func (c cmdable) Set(ctx context.Context, key string, value interface{}, expirat
 	}
 
 	cmd := NewStatusCmd(ctx, args...)
+	cmd.keys = append(cmd.keys, key)
 	_ = c(ctx, cmd)
 	return cmd
 }
@@ -230,6 +245,7 @@ func (c cmdable) SetArgs(ctx context.Context, key string, value interface{}, a S
 	}
 
 	cmd := NewStatusCmd(ctx, args...)
+	cmd.keys = append(cmd.keys, key)
 	_ = c(ctx, cmd)
 	return cmd
 }
@@ -237,6 +253,7 @@ func (c cmdable) SetArgs(ctx context.Context, key string, value interface{}, a S
 // SetEx Redis `SETEx key expiration value` command.
 func (c cmdable) SetEx(ctx context.Context, key string, value interface{}, expiration time.Duration) *StatusCmd {
 	cmd := NewStatusCmd(ctx, "setex", key, formatSec(ctx, expiration), value)
+	cmd.keys = append(cmd.keys, key)
 	_ = c(ctx, cmd)
 	return cmd
 }
@@ -261,7 +278,7 @@ func (c cmdable) SetNX(ctx context.Context, key string, value interface{}, expir
 			cmd = NewBoolCmd(ctx, "set", key, value, "ex", formatSec(ctx, expiration), "nx")
 		}
 	}
-
+	cmd.keys = append(cmd.keys, key)
 	_ = c(ctx, cmd)
 	return cmd
 }
@@ -285,19 +302,21 @@ func (c cmdable) SetXX(ctx context.Context, key string, value interface{}, expir
 			cmd = NewBoolCmd(ctx, "set", key, value, "ex", formatSec(ctx, expiration), "xx")
 		}
 	}
-
+	cmd.keys = append(cmd.keys, key)
 	_ = c(ctx, cmd)
 	return cmd
 }
 
 func (c cmdable) SetRange(ctx context.Context, key string, offset int64, value string) *IntCmd {
 	cmd := NewIntCmd(ctx, "setrange", key, offset, value)
+	cmd.keys = append(cmd.keys, key)
 	_ = c(ctx, cmd)
 	return cmd
 }
 
 func (c cmdable) StrLen(ctx context.Context, key string) *IntCmd {
 	cmd := NewIntCmd(ctx, "strlen", key)
+	cmd.keys = append(cmd.keys, key)
 	_ = c(ctx, cmd)
 	return cmd
 }

timeseries_commands_test.go

@@ -2,6 +2,7 @@ package redis_test
 
 import (
 	"context"
+	"fmt"
 	"strings"
 
 	. "github.com/bsm/ginkgo/v2"
@@ -12,15 +13,33 @@ import (
 
 var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
 	ctx := context.TODO()
+
+	setupRedisClient := func(protocolVersion int) *redis.Client {
+		return redis.NewClient(&redis.Options{
+			Addr:          "localhost:6379",
+			DB:            0,
+			Protocol:      protocolVersion,
+			UnstableResp3: true,
+		})
+	}
+
+	protocols := []int{2, 3}
+	for _, protocol := range protocols {
+		protocol := protocol // capture loop variable for each context
+
+		Context(fmt.Sprintf("with protocol version %d", protocol), func() {
 			var client *redis.Client
 
 			BeforeEach(func() {
-		client = redis.NewClient(&redis.Options{Addr: rediStackAddr})
-		Expect(client.FlushDB(ctx).Err()).NotTo(HaveOccurred())
+				client = setupRedisClient(protocol)
+				Expect(client.FlushAll(ctx).Err()).NotTo(HaveOccurred())
 			})
 
 			AfterEach(func() {
-		Expect(client.Close()).NotTo(HaveOccurred())
+				if client != nil {
+					client.FlushDB(ctx)
+					client.Close()
+				}
 			})
 
 			It("should TSCreate and TSCreateWithArgs", Label("timeseries", "tscreate", "tscreateWithArgs", "NonRedisEnterprise"), func() {
@@ -42,7 +61,11 @@ var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
 				Expect(result).To(BeEquivalentTo("OK"))
 				resultInfo, err := client.TSInfo(ctx, "4").Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(resultInfo["labels"].([]interface{})[0]).To(BeEquivalentTo([]interface{}{"Time", "Series"}))
+				} else {
 					Expect(resultInfo["labels"].(map[interface{}]interface{})["Time"]).To(BeEquivalentTo("Series"))
+				}
 				// Test chunk size
 				opt = &redis.TSOptions{ChunkSize: 128}
 				result, err = client.TSCreateWithArgs(ctx, "ts-cs-1", opt).Result()
@@ -134,7 +157,11 @@ var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
 				Expect(result).To(BeEquivalentTo(4))
 				resultInfo, err := client.TSInfo(ctx, "4").Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(resultInfo["labels"].([]interface{})).To(ContainElement([]interface{}{"Time", "Series"}))
+				} else {
 					Expect(resultInfo["labels"].(map[interface{}]interface{})["Time"]).To(BeEquivalentTo("Series"))
+				}
 				// Test chunk size
 				opt = &redis.TSOptions{ChunkSize: 128}
 				result, err = client.TSAddWithArgs(ctx, "ts-cs-1", 1, 10, opt).Result()
@@ -223,7 +250,11 @@ var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
 
 				resultInfo, err = client.TSInfo(ctx, "1").Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(resultInfo["labels"]).To(BeEquivalentTo([]interface{}{}))
+				} else {
 					Expect(resultInfo["labels"]).To(BeEquivalentTo(map[interface{}]interface{}{}))
+				}
 
 				opt = &redis.TSAlterOptions{Labels: map[string]string{"Time": "Series"}}
 				resultAlter, err = client.TSAlter(ctx, "1", opt).Result()
@@ -232,9 +263,15 @@ var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
 
 				resultInfo, err = client.TSInfo(ctx, "1").Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(resultInfo["labels"].([]interface{})[0]).To(BeEquivalentTo([]interface{}{"Time", "Series"}))
+					Expect(resultInfo["retentionTime"]).To(BeEquivalentTo(10))
+					Expect(resultInfo["duplicatePolicy"]).To(BeEquivalentTo(redis.Nil))
+				} else {
 					Expect(resultInfo["labels"].(map[interface{}]interface{})["Time"]).To(BeEquivalentTo("Series"))
 					Expect(resultInfo["retentionTime"]).To(BeEquivalentTo(10))
 					Expect(resultInfo["duplicatePolicy"]).To(BeEquivalentTo(redis.Nil))
+				}
 				opt = &redis.TSAlterOptions{DuplicatePolicy: "min"}
 				resultAlter, err = client.TSAlter(ctx, "1", opt).Result()
 				Expect(err).NotTo(HaveOccurred())
@@ -304,7 +341,11 @@ var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
 				Expect(resultDeleteRule).To(BeEquivalentTo("OK"))
 				resultInfo, err := client.TSInfo(ctx, "1").Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(resultInfo["rules"]).To(BeEquivalentTo([]interface{}{}))
+				} else {
 					Expect(resultInfo["rules"]).To(BeEquivalentTo(map[interface{}]interface{}{}))
+				}
 			})
 
 			It("should TSIncrBy, TSIncrByWithArgs, TSDecrBy and TSDecrByWithArgs", Label("timeseries", "tsincrby", "tsdecrby", "tsincrbyWithArgs", "tsdecrbyWithArgs", "NonRedisEnterprise"), func() {
@@ -501,12 +542,21 @@ var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
 
 				result, err := client.TSMGet(ctx, []string{"Test=This"}).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(result["a"][1].([]interface{})[1]).To(BeEquivalentTo("15"))
+					Expect(result["b"][1].([]interface{})[1]).To(BeEquivalentTo("25"))
+				} else {
 					Expect(result["a"][1].([]interface{})[1]).To(BeEquivalentTo(15))
 					Expect(result["b"][1].([]interface{})[1]).To(BeEquivalentTo(25))
+				}
 				mgetOpt := &redis.TSMGetOptions{WithLabels: true}
 				result, err = client.TSMGetWithArgs(ctx, []string{"Test=This"}, mgetOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(result["b"][0]).To(ConsistOf([]interface{}{"Test", "This"}, []interface{}{"Taste", "That"}))
+				} else {
 					Expect(result["b"][0]).To(BeEquivalentTo(map[interface{}]interface{}{"Test": "This", "Taste": "That"}))
+				}
 
 				resultCreate, err = client.TSCreate(ctx, "c").Result()
 				Expect(err).NotTo(HaveOccurred())
@@ -528,11 +578,19 @@ var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
 				Expect(err).NotTo(HaveOccurred())
 				result, err = client.TSMGet(ctx, []string{"is_compaction=true"}).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(result["d"][1]).To(BeEquivalentTo([]interface{}{int64(0), "4"}))
+				} else {
 					Expect(result["d"][1]).To(BeEquivalentTo([]interface{}{int64(0), 4.0}))
+				}
 				mgetOpt = &redis.TSMGetOptions{Latest: true}
 				result, err = client.TSMGetWithArgs(ctx, []string{"is_compaction=true"}, mgetOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(result["d"][1]).To(BeEquivalentTo([]interface{}{int64(10), "8"}))
+				} else {
 					Expect(result["d"][1]).To(BeEquivalentTo([]interface{}{int64(10), 8.0}))
+				}
 			})
 
 			It("should TSQueryIndex", Label("timeseries", "tsqueryindex"), func() {
@@ -830,12 +888,20 @@ var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
 				result, err := client.TSMRange(ctx, 0, 200, []string{"Test=This"}).Result()
 				Expect(err).NotTo(HaveOccurred())
 				Expect(len(result)).To(BeEquivalentTo(2))
+				if client.Options().Protocol == 2 {
+					Expect(len(result["a"][1].([]interface{}))).To(BeEquivalentTo(100))
+				} else {
 					Expect(len(result["a"][2].([]interface{}))).To(BeEquivalentTo(100))
+				}
 				// Test Count
 				mrangeOpt := &redis.TSMRangeOptions{Count: 10}
 				result, err = client.TSMRangeWithArgs(ctx, 0, 200, []string{"Test=This"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(len(result["a"][1].([]interface{}))).To(BeEquivalentTo(10))
+				} else {
 					Expect(len(result["a"][2].([]interface{}))).To(BeEquivalentTo(10))
+				}
 				// Test Aggregation and BucketDuration
 				for i := 0; i < 100; i++ {
 					_, err := client.TSAdd(ctx, "a", i+200, float64(i%7)).Result()
@@ -845,19 +911,36 @@ var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
 				result, err = client.TSMRangeWithArgs(ctx, 0, 500, []string{"Test=This"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
 				Expect(len(result)).To(BeEquivalentTo(2))
+				if client.Options().Protocol == 2 {
+					Expect(len(result["a"][1].([]interface{}))).To(BeEquivalentTo(20))
+				} else {
 					Expect(len(result["a"][2].([]interface{}))).To(BeEquivalentTo(20))
+				}
 				// Test WithLabels
+				if client.Options().Protocol == 2 {
+					Expect(result["a"][0]).To(BeEquivalentTo([]interface{}{}))
+				} else {
 					Expect(result["a"][0]).To(BeEquivalentTo(map[interface{}]interface{}{}))
+				}
 				mrangeOpt = &redis.TSMRangeOptions{WithLabels: true}
 				result, err = client.TSMRangeWithArgs(ctx, 0, 200, []string{"Test=This"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(result["a"][0]).To(ConsistOf([]interface{}{[]interface{}{"Test", "This"}, []interface{}{"team", "ny"}}))
+				} else {
 					Expect(result["a"][0]).To(BeEquivalentTo(map[interface{}]interface{}{"Test": "This", "team": "ny"}))
+				}
 				// Test SelectedLabels
 				mrangeOpt = &redis.TSMRangeOptions{SelectedLabels: []interface{}{"team"}}
 				result, err = client.TSMRangeWithArgs(ctx, 0, 200, []string{"Test=This"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(result["a"][0].([]interface{})[0]).To(BeEquivalentTo([]interface{}{"team", "ny"}))
+					Expect(result["b"][0].([]interface{})[0]).To(BeEquivalentTo([]interface{}{"team", "sf"}))
+				} else {
 					Expect(result["a"][0]).To(BeEquivalentTo(map[interface{}]interface{}{"team": "ny"}))
 					Expect(result["b"][0]).To(BeEquivalentTo(map[interface{}]interface{}{"team": "sf"}))
+				}
 				// Test FilterBy
 				fts := make([]int, 0)
 				for i := 10; i < 20; i++ {
@@ -866,34 +949,58 @@ var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
 				mrangeOpt = &redis.TSMRangeOptions{FilterByTS: fts, FilterByValue: []int{1, 2}}
 				result, err = client.TSMRangeWithArgs(ctx, 0, 200, []string{"Test=This"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(result["a"][1].([]interface{})).To(BeEquivalentTo([]interface{}{[]interface{}{int64(15), "1"}, []interface{}{int64(16), "2"}}))
+				} else {
 					Expect(result["a"][2]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(15), 1.0}, []interface{}{int64(16), 2.0}}))
+				}
 				// Test GroupBy
 				mrangeOpt = &redis.TSMRangeOptions{GroupByLabel: "Test", Reducer: "sum"}
 				result, err = client.TSMRangeWithArgs(ctx, 0, 3, []string{"Test=This"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(result["Test=This"][1]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(0), "0"}, []interface{}{int64(1), "2"}, []interface{}{int64(2), "4"}, []interface{}{int64(3), "6"}}))
+				} else {
 					Expect(result["Test=This"][3]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(0), 0.0}, []interface{}{int64(1), 2.0}, []interface{}{int64(2), 4.0}, []interface{}{int64(3), 6.0}}))
-
+				}
 				mrangeOpt = &redis.TSMRangeOptions{GroupByLabel: "Test", Reducer: "max"}
 				result, err = client.TSMRangeWithArgs(ctx, 0, 3, []string{"Test=This"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(result["Test=This"][1]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(0), "0"}, []interface{}{int64(1), "1"}, []interface{}{int64(2), "2"}, []interface{}{int64(3), "3"}}))
+				} else {
 					Expect(result["Test=This"][3]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(0), 0.0}, []interface{}{int64(1), 1.0}, []interface{}{int64(2), 2.0}, []interface{}{int64(3), 3.0}}))
+				}
 
 				mrangeOpt = &redis.TSMRangeOptions{GroupByLabel: "team", Reducer: "min"}
 				result, err = client.TSMRangeWithArgs(ctx, 0, 3, []string{"Test=This"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
 				Expect(len(result)).To(BeEquivalentTo(2))
+				if client.Options().Protocol == 2 {
+					Expect(result["team=ny"][1]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(0), "0"}, []interface{}{int64(1), "1"}, []interface{}{int64(2), "2"}, []interface{}{int64(3), "3"}}))
+					Expect(result["team=sf"][1]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(0), "0"}, []interface{}{int64(1), "1"}, []interface{}{int64(2), "2"}, []interface{}{int64(3), "3"}}))
+				} else {
 					Expect(result["team=ny"][3]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(0), 0.0}, []interface{}{int64(1), 1.0}, []interface{}{int64(2), 2.0}, []interface{}{int64(3), 3.0}}))
 					Expect(result["team=sf"][3]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(0), 0.0}, []interface{}{int64(1), 1.0}, []interface{}{int64(2), 2.0}, []interface{}{int64(3), 3.0}}))
+				}
 				// Test Align
 				mrangeOpt = &redis.TSMRangeOptions{Aggregator: redis.Count, BucketDuration: 10, Align: "-"}
 				result, err = client.TSMRangeWithArgs(ctx, 0, 10, []string{"team=ny"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(result["a"][1]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(0), "10"}, []interface{}{int64(10), "1"}}))
+				} else {
 					Expect(result["a"][2]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(0), 10.0}, []interface{}{int64(10), 1.0}}))
+				}
 
 				mrangeOpt = &redis.TSMRangeOptions{Aggregator: redis.Count, BucketDuration: 10, Align: 5}
 				result, err = client.TSMRangeWithArgs(ctx, 0, 10, []string{"team=ny"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(result["a"][1]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(0), "5"}, []interface{}{int64(5), "6"}}))
+				} else {
 					Expect(result["a"][2]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(0), 5.0}, []interface{}{int64(5), 6.0}}))
+				}
 			})
 
 			It("should TSMRangeWithArgs Latest", Label("timeseries", "tsmrangeWithArgs", "tsmrangelatest", "NonRedisEnterprise"), func() {
@@ -940,8 +1047,13 @@ var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
 				mrangeOpt := &redis.TSMRangeOptions{Latest: true}
 				result, err := client.TSMRangeWithArgs(ctx, 0, 10, []string{"is_compaction=true"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(result["b"][1]).To(ConsistOf([]interface{}{int64(0), "4"}, []interface{}{int64(10), "8"}))
+					Expect(result["d"][1]).To(ConsistOf([]interface{}{int64(0), "4"}, []interface{}{int64(10), "8"}))
+				} else {
 					Expect(result["b"][2]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(0), 4.0}, []interface{}{int64(10), 8.0}}))
 					Expect(result["d"][2]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(0), 4.0}, []interface{}{int64(10), 8.0}}))
+				}
 			})
 			It("should TSMRevRange and TSMRevRangeWithArgs", Label("timeseries", "tsmrevrange", "tsmrevrangeWithArgs"), func() {
 				createOpt := &redis.TSOptions{Labels: map[string]string{"Test": "This", "team": "ny"}}
@@ -962,12 +1074,20 @@ var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
 				result, err := client.TSMRevRange(ctx, 0, 200, []string{"Test=This"}).Result()
 				Expect(err).NotTo(HaveOccurred())
 				Expect(len(result)).To(BeEquivalentTo(2))
+				if client.Options().Protocol == 2 {
+					Expect(len(result["a"][1].([]interface{}))).To(BeEquivalentTo(100))
+				} else {
 					Expect(len(result["a"][2].([]interface{}))).To(BeEquivalentTo(100))
+				}
 				// Test Count
 				mrangeOpt := &redis.TSMRevRangeOptions{Count: 10}
 				result, err = client.TSMRevRangeWithArgs(ctx, 0, 200, []string{"Test=This"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(len(result["a"][1].([]interface{}))).To(BeEquivalentTo(10))
+				} else {
 					Expect(len(result["a"][2].([]interface{}))).To(BeEquivalentTo(10))
+				}
 				// Test Aggregation and BucketDuration
 				for i := 0; i < 100; i++ {
 					_, err := client.TSAdd(ctx, "a", i+200, float64(i%7)).Result()
@@ -977,20 +1097,32 @@ var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
 				result, err = client.TSMRevRangeWithArgs(ctx, 0, 500, []string{"Test=This"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
 				Expect(len(result)).To(BeEquivalentTo(2))
+				if client.Options().Protocol == 2 {
+					Expect(len(result["a"][1].([]interface{}))).To(BeEquivalentTo(20))
+					Expect(result["a"][0]).To(BeEquivalentTo([]interface{}{}))
+				} else {
 					Expect(len(result["a"][2].([]interface{}))).To(BeEquivalentTo(20))
 					Expect(result["a"][0]).To(BeEquivalentTo(map[interface{}]interface{}{}))
-		// Test WithLabels
-		Expect(result["a"][0]).To(BeEquivalentTo(map[interface{}]interface{}{}))
+				}
 				mrangeOpt = &redis.TSMRevRangeOptions{WithLabels: true}
 				result, err = client.TSMRevRangeWithArgs(ctx, 0, 200, []string{"Test=This"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(result["a"][0]).To(ConsistOf([]interface{}{[]interface{}{"Test", "This"}, []interface{}{"team", "ny"}}))
+				} else {
 					Expect(result["a"][0]).To(BeEquivalentTo(map[interface{}]interface{}{"Test": "This", "team": "ny"}))
+				}
 				// Test SelectedLabels
 				mrangeOpt = &redis.TSMRevRangeOptions{SelectedLabels: []interface{}{"team"}}
 				result, err = client.TSMRevRangeWithArgs(ctx, 0, 200, []string{"Test=This"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(result["a"][0].([]interface{})[0]).To(BeEquivalentTo([]interface{}{"team", "ny"}))
+					Expect(result["b"][0].([]interface{})[0]).To(BeEquivalentTo([]interface{}{"team", "sf"}))
+				} else {
 					Expect(result["a"][0]).To(BeEquivalentTo(map[interface{}]interface{}{"team": "ny"}))
 					Expect(result["b"][0]).To(BeEquivalentTo(map[interface{}]interface{}{"team": "sf"}))
+				}
 				// Test FilterBy
 				fts := make([]int, 0)
 				for i := 10; i < 20; i++ {
@@ -999,34 +1131,56 @@ var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
 				mrangeOpt = &redis.TSMRevRangeOptions{FilterByTS: fts, FilterByValue: []int{1, 2}}
 				result, err = client.TSMRevRangeWithArgs(ctx, 0, 200, []string{"Test=This"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(result["a"][1].([]interface{})).To(ConsistOf([]interface{}{int64(16), "2"}, []interface{}{int64(15), "1"}))
+				} else {
 					Expect(result["a"][2]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(16), 2.0}, []interface{}{int64(15), 1.0}}))
+				}
 				// Test GroupBy
 				mrangeOpt = &redis.TSMRevRangeOptions{GroupByLabel: "Test", Reducer: "sum"}
 				result, err = client.TSMRevRangeWithArgs(ctx, 0, 3, []string{"Test=This"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(result["Test=This"][1]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(3), "6"}, []interface{}{int64(2), "4"}, []interface{}{int64(1), "2"}, []interface{}{int64(0), "0"}}))
+				} else {
 					Expect(result["Test=This"][3]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(3), 6.0}, []interface{}{int64(2), 4.0}, []interface{}{int64(1), 2.0}, []interface{}{int64(0), 0.0}}))
-
+				}
 				mrangeOpt = &redis.TSMRevRangeOptions{GroupByLabel: "Test", Reducer: "max"}
 				result, err = client.TSMRevRangeWithArgs(ctx, 0, 3, []string{"Test=This"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(result["Test=This"][1]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(3), "3"}, []interface{}{int64(2), "2"}, []interface{}{int64(1), "1"}, []interface{}{int64(0), "0"}}))
+				} else {
 					Expect(result["Test=This"][3]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(3), 3.0}, []interface{}{int64(2), 2.0}, []interface{}{int64(1), 1.0}, []interface{}{int64(0), 0.0}}))
-
+				}
 				mrangeOpt = &redis.TSMRevRangeOptions{GroupByLabel: "team", Reducer: "min"}
 				result, err = client.TSMRevRangeWithArgs(ctx, 0, 3, []string{"Test=This"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
 				Expect(len(result)).To(BeEquivalentTo(2))
+				if client.Options().Protocol == 2 {
+					Expect(result["team=ny"][1]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(3), "3"}, []interface{}{int64(2), "2"}, []interface{}{int64(1), "1"}, []interface{}{int64(0), "0"}}))
+					Expect(result["team=sf"][1]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(3), "3"}, []interface{}{int64(2), "2"}, []interface{}{int64(1), "1"}, []interface{}{int64(0), "0"}}))
+				} else {
 					Expect(result["team=ny"][3]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(3), 3.0}, []interface{}{int64(2), 2.0}, []interface{}{int64(1), 1.0}, []interface{}{int64(0), 0.0}}))
 					Expect(result["team=sf"][3]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(3), 3.0}, []interface{}{int64(2), 2.0}, []interface{}{int64(1), 1.0}, []interface{}{int64(0), 0.0}}))
+				}
 				// Test Align
 				mrangeOpt = &redis.TSMRevRangeOptions{Aggregator: redis.Count, BucketDuration: 10, Align: "-"}
 				result, err = client.TSMRevRangeWithArgs(ctx, 0, 10, []string{"team=ny"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(result["a"][1]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(10), "1"}, []interface{}{int64(0), "10"}}))
+				} else {
 					Expect(result["a"][2]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(10), 1.0}, []interface{}{int64(0), 10.0}}))
-
+				}
 				mrangeOpt = &redis.TSMRevRangeOptions{Aggregator: redis.Count, BucketDuration: 10, Align: 1}
 				result, err = client.TSMRevRangeWithArgs(ctx, 0, 10, []string{"team=ny"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(result["a"][1]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(1), "10"}, []interface{}{int64(0), "1"}}))
+				} else {
 					Expect(result["a"][2]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(1), 10.0}, []interface{}{int64(0), 1.0}}))
+				}
 			})
 
 			It("should TSMRevRangeWithArgs Latest", Label("timeseries", "tsmrevrangeWithArgs", "tsmrevrangelatest", "NonRedisEnterprise"), func() {
@@ -1073,7 +1227,14 @@ var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
 				mrangeOpt := &redis.TSMRevRangeOptions{Latest: true}
 				result, err := client.TSMRevRangeWithArgs(ctx, 0, 10, []string{"is_compaction=true"}, mrangeOpt).Result()
 				Expect(err).NotTo(HaveOccurred())
+				if client.Options().Protocol == 2 {
+					Expect(result["b"][1]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(10), "8"}, []interface{}{int64(0), "4"}}))
+					Expect(result["d"][1]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(10), "8"}, []interface{}{int64(0), "4"}}))
+				} else {
 					Expect(result["b"][2]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(10), 8.0}, []interface{}{int64(0), 4.0}}))
 					Expect(result["d"][2]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(10), 8.0}, []interface{}{int64(0), 4.0}}))
+				}
 			})
+		})
+	}
 })