Compare commits

..

1 Commits

Author SHA1 Message Date
Flc゛ ca352ceb1d
Merge cc4810cc5a into 80c9f5bb77 2024-11-08 13:07:25 +01:00
6 changed files with 1802 additions and 2068 deletions

View File

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

View File

@ -186,21 +186,6 @@ 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!

View File

@ -1403,18 +1403,11 @@ 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 {
@ -1433,35 +1426,6 @@ 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
}

View File

@ -319,68 +319,36 @@ 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
}
for i := 0; i < n; i++ {
key, err := rd.ReadString()
var key string
var result BFInfo
for f := 0; f < n; f++ {
key, err = rd.ReadString()
if err != nil {
return err
}
if err := readAndAssignValue(key); 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 {
return err
}
}

View File

@ -13,32 +13,15 @@ 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 = setupRedisClient(protocol)
Expect(client.FlushAll(ctx).Err()).NotTo(HaveOccurred())
client = redis.NewClient(&redis.Options{Addr: ":6379"})
Expect(client.FlushDB(ctx).Err()).NotTo(HaveOccurred())
})
AfterEach(func() {
if client != nil {
client.FlushDB(ctx)
client.Close()
}
Expect(client.Close()).NotTo(HaveOccurred())
})
Describe("bloom", Label("bloom"), func() {
@ -134,7 +117,7 @@ var _ = Describe("Probabilistic commands", Label("probabilistic"), func() {
NoCreate: true,
}
_, err := client.BFInsert(ctx, "testbf1", options, "item1").Result()
resultInsert, err := client.BFInsert(ctx, "testbf1", options, "item1").Result()
Expect(err).To(HaveOccurred())
Expect(err).To(MatchError("ERR not found"))
@ -146,7 +129,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))
@ -413,7 +396,7 @@ var _ = Describe("Probabilistic commands", Label("probabilistic"), func() {
NoCreate: true,
}
_, err := client.CFInsertNX(ctx, "testcf1", args, "item1", "item2", "item2").Result()
result, err := client.CFInsertNX(ctx, "testcf1", args, "item1", "item2", "item2").Result()
Expect(err).To(HaveOccurred())
args = &redis.CFInsertOptions{
@ -421,7 +404,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)))
@ -747,6 +730,4 @@ var _ = Describe("Probabilistic commands", Label("probabilistic"), func() {
Expect(max).To(BeEquivalentTo(float64(140)))
})
})
})
}
})

View File

@ -2,7 +2,6 @@ package redis_test
import (
"context"
"fmt"
"strings"
. "github.com/bsm/ginkgo/v2"
@ -13,33 +12,15 @@ 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 = setupRedisClient(protocol)
Expect(client.FlushAll(ctx).Err()).NotTo(HaveOccurred())
client = redis.NewClient(&redis.Options{Addr: rediStackAddr})
Expect(client.FlushDB(ctx).Err()).NotTo(HaveOccurred())
})
AfterEach(func() {
if client != nil {
client.FlushDB(ctx)
client.Close()
}
Expect(client.Close()).NotTo(HaveOccurred())
})
It("should TSCreate and TSCreateWithArgs", Label("timeseries", "tscreate", "tscreateWithArgs", "NonRedisEnterprise"), func() {
@ -61,11 +42,7 @@ 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()
@ -157,11 +134,7 @@ 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()
@ -250,11 +223,7 @@ 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()
@ -263,15 +232,9 @@ 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())
@ -341,11 +304,7 @@ 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() {
@ -542,21 +501,12 @@ 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())
@ -578,19 +528,11 @@ 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() {
@ -888,20 +830,12 @@ 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()
@ -911,36 +845,19 @@ 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++ {
@ -949,58 +866,34 @@ 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() {
@ -1047,13 +940,8 @@ 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"}}
@ -1074,20 +962,12 @@ 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()
@ -1097,32 +977,20 @@ 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++ {
@ -1131,56 +999,34 @@ 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() {
@ -1227,14 +1073,7 @@ 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}}))
}
})
})
}
})