From 558581eac6ee7527407045859542a118157efab4 Mon Sep 17 00:00:00 2001 From: ofekshenawa <104765379+ofekshenawa@users.noreply.github.com> Date: Tue, 15 Aug 2023 10:19:19 +0300 Subject: [PATCH] bloom support (#2673) --- .github/wordlist.txt | 1 + README.md | 1 + commands.go | 2 + probabilistic.go | 1433 +++++++++++++++++++++++++++++++++++++++++ probabilistic_test.go | 739 +++++++++++++++++++++ 5 files changed, 2176 insertions(+) create mode 100644 probabilistic.go create mode 100644 probabilistic_test.go diff --git a/.github/wordlist.txt b/.github/wordlist.txt index 3045ae93..294988d5 100644 --- a/.github/wordlist.txt +++ b/.github/wordlist.txt @@ -52,3 +52,4 @@ uri URI url variadic +RedisStack \ No newline at end of file diff --git a/README.md b/README.md index 55dbec51..3486e8e5 100644 --- a/README.md +++ b/README.md @@ -46,6 +46,7 @@ key value NoSQL database that uses RocksDB as storage engine and is compatible w - [Redis Cluster](https://redis.uptrace.dev/guide/go-redis-cluster.html). - [Redis Ring](https://redis.uptrace.dev/guide/ring.html). - [Redis Performance Monitoring](https://redis.uptrace.dev/guide/redis-performance-monitoring.html). +- [Redis Probabilistic [RedisStack]](https://redis.io/docs/data-types/probabilistic/) ## Installation diff --git a/commands.go b/commands.go index 34f4d2c2..41f8d3ae 100644 --- a/commands.go +++ b/commands.go @@ -504,6 +504,8 @@ type Cmdable interface { ACLLogReset(ctx context.Context) *StatusCmd ModuleLoadex(ctx context.Context, conf *ModuleLoadexConfig) *StringCmd + + probabilisticCmdable } type StatefulCmdable interface { diff --git a/probabilistic.go b/probabilistic.go new file mode 100644 index 00000000..8e32bca9 --- /dev/null +++ b/probabilistic.go @@ -0,0 +1,1433 @@ +package redis + +import ( + "context" + "fmt" + + "github.com/redis/go-redis/v9/internal/proto" +) + +type probabilisticCmdable interface { + BFAdd(ctx context.Context, key string, element interface{}) *BoolCmd + BFCard(ctx context.Context, key string) *IntCmd + BFExists(ctx context.Context, key string, element interface{}) *BoolCmd + BFInfo(ctx context.Context, key string) *BFInfoCmd + BFInfoArg(ctx context.Context, key, option string) *BFInfoCmd + BFInfoCapacity(ctx context.Context, key string) *BFInfoCmd + BFInfoSize(ctx context.Context, key string) *BFInfoCmd + BFInfoFilters(ctx context.Context, key string) *BFInfoCmd + BFInfoItems(ctx context.Context, key string) *BFInfoCmd + BFInfoExpansion(ctx context.Context, key string) *BFInfoCmd + BFInsert(ctx context.Context, key string, options *BFInsertOptions, elements ...interface{}) *BoolSliceCmd + BFMAdd(ctx context.Context, key string, elements ...interface{}) *BoolSliceCmd + BFMExists(ctx context.Context, key string, elements ...interface{}) *BoolSliceCmd + BFReserve(ctx context.Context, key string, errorRate float64, capacity int64) *StatusCmd + BFReserveExpansion(ctx context.Context, key string, errorRate float64, capacity, expansion int64) *StatusCmd + BFReserveNonScaling(ctx context.Context, key string, errorRate float64, capacity int64) *StatusCmd + BFReserveArgs(ctx context.Context, key string, options *BFReserveOptions) *StatusCmd + BFScanDump(ctx context.Context, key string, iterator int64) *ScanDumpCmd + BFLoadChunk(ctx context.Context, key string, iterator int64, data interface{}) *StatusCmd + + CFAdd(ctx context.Context, key string, element interface{}) *BoolCmd + CFAddNX(ctx context.Context, key string, element interface{}) *BoolCmd + CFCount(ctx context.Context, key string, element interface{}) *IntCmd + CFDel(ctx context.Context, key string, element interface{}) *BoolCmd + CFExists(ctx context.Context, key string, element interface{}) *BoolCmd + CFInfo(ctx context.Context, key string) *CFInfoCmd + CFInsert(ctx context.Context, key string, options *CFInsertOptions, elements ...interface{}) *BoolSliceCmd + CFInsertNX(ctx context.Context, key string, options *CFInsertOptions, elements ...interface{}) *IntSliceCmd + CFMExists(ctx context.Context, key string, elements ...interface{}) *BoolSliceCmd + CFReserve(ctx context.Context, key string, capacity int64) *StatusCmd + CFReserveArgs(ctx context.Context, key string, options *CFReserveOptions) *StatusCmd + CFReserveExpansion(ctx context.Context, key string, capacity int64, expansion int64) *StatusCmd + CFReserveBucketSize(ctx context.Context, key string, capacity int64, bucketsize int64) *StatusCmd + CFReserveMaxIterations(ctx context.Context, key string, capacity int64, maxiterations int64) *StatusCmd + CFScanDump(ctx context.Context, key string, iterator int64) *ScanDumpCmd + CFLoadChunk(ctx context.Context, key string, iterator int64, data interface{}) *StatusCmd + + CMSIncrBy(ctx context.Context, key string, elements ...interface{}) *IntSliceCmd + CMSInfo(ctx context.Context, key string) *CMSInfoCmd + CMSInitByDim(ctx context.Context, key string, width, height int64) *StatusCmd + CMSInitByProb(ctx context.Context, key string, errorRate, probability float64) *StatusCmd + CMSMerge(ctx context.Context, destKey string, sourceKeys ...string) *StatusCmd + CMSMergeWithWeight(ctx context.Context, destKey string, sourceKeys map[string]int64) *StatusCmd + CMSQuery(ctx context.Context, key string, elements ...interface{}) *IntSliceCmd + + TopKAdd(ctx context.Context, key string, elements ...interface{}) *StringSliceCmd + TopKCount(ctx context.Context, key string, elements ...interface{}) *IntSliceCmd + TopKIncrBy(ctx context.Context, key string, elements ...interface{}) *StringSliceCmd + TopKInfo(ctx context.Context, key string) *TopKInfoCmd + TopKList(ctx context.Context, key string) *StringSliceCmd + TopKListWithCount(ctx context.Context, key string) *MapStringIntCmd + TopKQuery(ctx context.Context, key string, elements ...interface{}) *BoolSliceCmd + TopKReserve(ctx context.Context, key string, k int64) *StatusCmd + TopKReserveWithOptions(ctx context.Context, key string, k int64, width, depth int64, decay float64) *StatusCmd + + TDigestAdd(ctx context.Context, key string, elements ...float64) *StatusCmd + TDigestByRank(ctx context.Context, key string, rank ...uint64) *FloatSliceCmd + TDigestByRevRank(ctx context.Context, key string, rank ...uint64) *FloatSliceCmd + TDigestCDF(ctx context.Context, key string, elements ...float64) *FloatSliceCmd + TDigestCreate(ctx context.Context, key string) *StatusCmd + TDigestCreateWithCompression(ctx context.Context, key string, compression int64) *StatusCmd + TDigestInfo(ctx context.Context, key string) *TDigestInfoCmd + TDigestMax(ctx context.Context, key string) *FloatCmd + TDigestMin(ctx context.Context, key string) *FloatCmd + TDigestMerge(ctx context.Context, destKey string, options *TDigestMergeOptions, sourceKeys ...string) *StatusCmd + TDigestQuantile(ctx context.Context, key string, elements ...float64) *FloatSliceCmd + TDigestRank(ctx context.Context, key string, values ...float64) *IntSliceCmd + TDigestReset(ctx context.Context, key string) *StatusCmd + TDigestRevRank(ctx context.Context, key string, values ...float64) *IntSliceCmd + TDigestTrimmedMean(ctx context.Context, key string, lowCutQuantile, highCutQuantile float64) *FloatCmd +} + +type BFInsertOptions struct { + Capacity int64 + Error float64 + Expansion int64 + NonScaling bool + NoCreate bool +} + +type BFReserveOptions struct { + Capacity int64 + Error float64 + Expansion int64 + NonScaling bool +} + +type CFReserveOptions struct { + Capacity int64 + BucketSize int64 + MaxIterations int64 + Expansion int64 +} + +type CFInsertOptions struct { + Capacity int64 + NoCreate bool +} + +// ------------------------------------------- +// Bloom filter commands +//------------------------------------------- + +// BFReserve creates an empty Bloom filter with a single sub-filter +// for the initial specified capacity and with an upper bound error_rate. +// For more information - https://redis.io/commands/bf.reserve/ +func (c cmdable) BFReserve(ctx context.Context, key string, errorRate float64, capacity int64) *StatusCmd { + args := []interface{}{"BF.RESERVE", key, errorRate, capacity} + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// BFReserveExpansion creates an empty Bloom filter with a single sub-filter +// for the initial specified capacity and with an upper bound error_rate. +// This function also allows for specifying an expansion rate for the filter. +// For more information - https://redis.io/commands/bf.reserve/ +func (c cmdable) BFReserveExpansion(ctx context.Context, key string, errorRate float64, capacity, expansion int64) *StatusCmd { + args := []interface{}{"BF.RESERVE", key, errorRate, capacity, "EXPANSION", expansion} + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// BFReserveNonScaling creates an empty Bloom filter with a single sub-filter +// for the initial specified capacity and with an upper bound error_rate. +// This function also allows for specifying that the filter should not scale. +// For more information - https://redis.io/commands/bf.reserve/ +func (c cmdable) BFReserveNonScaling(ctx context.Context, key string, errorRate float64, capacity int64) *StatusCmd { + args := []interface{}{"BF.RESERVE", key, errorRate, capacity, "NONSCALING"} + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// BFReserveArgs creates an empty Bloom filter with a single sub-filter +// for the initial specified capacity and with an upper bound error_rate. +// This function also allows for specifying additional options such as expansion rate and non-scaling behavior. +// For more information - https://redis.io/commands/bf.reserve/ +func (c cmdable) BFReserveArgs(ctx context.Context, key string, options *BFReserveOptions) *StatusCmd { + args := []interface{}{"BF.RESERVE", key} + if options != nil { + if options.Error != 0 { + args = append(args, options.Error) + } + if options.Capacity != 0 { + args = append(args, options.Capacity) + } + if options.Expansion != 0 { + args = append(args, "EXPANSION", options.Expansion) + } + if options.NonScaling { + args = append(args, "NONSCALING") + } + } + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// BFAdd adds an item to a Bloom filter. +// For more information - https://redis.io/commands/bf.add/ +func (c cmdable) BFAdd(ctx context.Context, key string, element interface{}) *BoolCmd { + args := []interface{}{"BF.ADD", key, element} + cmd := NewBoolCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// BFCard returns the cardinality of a Bloom filter - +// number of items that were added to a Bloom filter and detected as unique +// (items that caused at least one bit to be set in at least one sub-filter). +// For more information - https://redis.io/commands/bf.card/ +func (c cmdable) BFCard(ctx context.Context, key string) *IntCmd { + args := []interface{}{"BF.CARD", key} + cmd := NewIntCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// BFExists determines whether a given item was added to a Bloom filter. +// For more information - https://redis.io/commands/bf.exists/ +func (c cmdable) BFExists(ctx context.Context, key string, element interface{}) *BoolCmd { + args := []interface{}{"BF.EXISTS", key, element} + cmd := NewBoolCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// BFLoadChunk restores a Bloom filter previously saved using BF.SCANDUMP. +// For more information - https://redis.io/commands/bf.loadchunk/ +func (c cmdable) BFLoadChunk(ctx context.Context, key string, iterator int64, data interface{}) *StatusCmd { + args := []interface{}{"BF.LOADCHUNK", key, iterator, data} + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// Begins an incremental save of the Bloom filter. +// This command is useful for large Bloom filters that cannot fit into the DUMP and RESTORE model. +// For more information - https://redis.io/commands/bf.scandump/ +func (c cmdable) BFScanDump(ctx context.Context, key string, iterator int64) *ScanDumpCmd { + args := []interface{}{"BF.SCANDUMP", key, iterator} + cmd := newScanDumpCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +type ScanDump struct { + Iter int64 + Data string +} + +type ScanDumpCmd struct { + baseCmd + + val ScanDump +} + +func newScanDumpCmd(ctx context.Context, args ...interface{}) *ScanDumpCmd { + return &ScanDumpCmd{ + baseCmd: baseCmd{ + ctx: ctx, + args: args, + }, + } +} + +func (cmd *ScanDumpCmd) String() string { + return cmdString(cmd, cmd.val) +} + +func (cmd *ScanDumpCmd) SetVal(val ScanDump) { + cmd.val = val +} + +func (cmd *ScanDumpCmd) Result() (ScanDump, error) { + return cmd.val, cmd.err +} + +func (cmd *ScanDumpCmd) Val() ScanDump { + return cmd.val +} + +func (cmd *ScanDumpCmd) readReply(rd *proto.Reader) (err error) { + n, err := rd.ReadMapLen() + if err != nil { + return err + } + cmd.val = ScanDump{} + for i := 0; i < n; i++ { + iter, err := rd.ReadInt() + if err != nil { + return err + } + data, err := rd.ReadString() + if err != nil { + return err + } + cmd.val.Data = data + cmd.val.Iter = iter + + } + + return nil +} + +// Returns information about a Bloom filter. +// For more information - https://redis.io/commands/bf.info/ +func (c cmdable) BFInfo(ctx context.Context, key string) *BFInfoCmd { + args := []interface{}{"BF.INFO", key} + cmd := NewBFInfoCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +type BFInfo struct { + Capacity int64 + Size int64 + Filters int64 + ItemsInserted int64 + ExpansionRate int64 +} + +type BFInfoCmd struct { + baseCmd + + val BFInfo +} + +func NewBFInfoCmd(ctx context.Context, args ...interface{}) *BFInfoCmd { + return &BFInfoCmd{ + baseCmd: baseCmd{ + ctx: ctx, + args: args, + }, + } +} + +func (cmd *BFInfoCmd) SetVal(val BFInfo) { + cmd.val = val +} +func (cmd *BFInfoCmd) String() string { + return cmdString(cmd, cmd.val) +} + +func (cmd *BFInfoCmd) Val() BFInfo { + return cmd.val +} + +func (cmd *BFInfoCmd) Result() (BFInfo, error) { + return cmd.val, cmd.err +} + +func (cmd *BFInfoCmd) readReply(rd *proto.Reader) (err error) { + 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() + 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 { + return err + } + } + + cmd.val = result + return nil +} + +// BFInfoCapacity returns information about the capacity of a Bloom filter. +// For more information - https://redis.io/commands/bf.info/ +func (c cmdable) BFInfoCapacity(ctx context.Context, key string) *BFInfoCmd { + return c.BFInfoArg(ctx, key, "CAPACITY") +} + +// BFInfoSize returns information about the size of a Bloom filter. +// For more information - https://redis.io/commands/bf.info/ +func (c cmdable) BFInfoSize(ctx context.Context, key string) *BFInfoCmd { + return c.BFInfoArg(ctx, key, "SIZE") +} + +// BFInfoFilters returns information about the filters of a Bloom filter. +// For more information - https://redis.io/commands/bf.info/ +func (c cmdable) BFInfoFilters(ctx context.Context, key string) *BFInfoCmd { + return c.BFInfoArg(ctx, key, "FILTERS") +} + +// BFInfoItems returns information about the items of a Bloom filter. +// For more information - https://redis.io/commands/bf.info/ +func (c cmdable) BFInfoItems(ctx context.Context, key string) *BFInfoCmd { + return c.BFInfoArg(ctx, key, "ITEMS") +} + +// BFInfoExpansion returns information about the expansion rate of a Bloom filter. +// For more information - https://redis.io/commands/bf.info/ +func (c cmdable) BFInfoExpansion(ctx context.Context, key string) *BFInfoCmd { + return c.BFInfoArg(ctx, key, "EXPANSION") +} + +// BFInfoArg returns information about a specific option of a Bloom filter. +// For more information - https://redis.io/commands/bf.info/ +func (c cmdable) BFInfoArg(ctx context.Context, key, option string) *BFInfoCmd { + args := []interface{}{"BF.INFO", key, option} + cmd := NewBFInfoCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// BFInsert inserts elements into a Bloom filter. +// This function also allows for specifying additional options such as: +// capacity, error rate, expansion rate, and non-scaling behavior. +// For more information - https://redis.io/commands/bf.insert/ +func (c cmdable) BFInsert(ctx context.Context, key string, options *BFInsertOptions, elements ...interface{}) *BoolSliceCmd { + args := []interface{}{"BF.INSERT", key} + if options != nil { + if options.Capacity != 0 { + args = append(args, "CAPACITY", options.Capacity) + } + if options.Error != 0 { + args = append(args, "ERROR", options.Error) + } + if options.Expansion != 0 { + args = append(args, "EXPANSION", options.Expansion) + } + if options.NoCreate { + args = append(args, "NOCREATE") + } + if options.NonScaling { + args = append(args, "NONSCALING") + } + } + args = append(args, "ITEMS") + args = append(args, elements...) + + cmd := NewBoolSliceCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// BFMAdd adds multiple elements to a Bloom filter. +// Returns an array of booleans indicating whether each element was added to the filter or not. +// For more information - https://redis.io/commands/bf.madd/ +func (c cmdable) BFMAdd(ctx context.Context, key string, elements ...interface{}) *BoolSliceCmd { + args := []interface{}{"BF.MADD", key} + args = append(args, elements...) + cmd := NewBoolSliceCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// BFMExists check if multiple elements exist in a Bloom filter. +// Returns an array of booleans indicating whether each element exists in the filter or not. +// For more information - https://redis.io/commands/bf.mexists/ +func (c cmdable) BFMExists(ctx context.Context, key string, elements ...interface{}) *BoolSliceCmd { + args := []interface{}{"BF.MEXISTS", key} + args = append(args, elements...) + + cmd := NewBoolSliceCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// ------------------------------------------- +// Cuckoo filter commands +//------------------------------------------- + +// CFReserve creates an empty Cuckoo filter with the specified capacity. +// For more information - https://redis.io/commands/cf.reserve/ +func (c cmdable) CFReserve(ctx context.Context, key string, capacity int64) *StatusCmd { + args := []interface{}{"CF.RESERVE", key, capacity} + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// CFReserveExpansion creates an empty Cuckoo filter with the specified capacity and expansion rate. +// For more information - https://redis.io/commands/cf.reserve/ +func (c cmdable) CFReserveExpansion(ctx context.Context, key string, capacity int64, expansion int64) *StatusCmd { + args := []interface{}{"CF.RESERVE", key, capacity, "EXPANSION", expansion} + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// CFReserveBucketSize creates an empty Cuckoo filter with the specified capacity and bucket size. +// For more information - https://redis.io/commands/cf.reserve/ +func (c cmdable) CFReserveBucketSize(ctx context.Context, key string, capacity int64, bucketsize int64) *StatusCmd { + args := []interface{}{"CF.RESERVE", key, capacity, "BUCKETSIZE", bucketsize} + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// CFReserveMaxIterations creates an empty Cuckoo filter with the specified capacity and maximum number of iterations. +// For more information - https://redis.io/commands/cf.reserve/ +func (c cmdable) CFReserveMaxIterations(ctx context.Context, key string, capacity int64, maxiterations int64) *StatusCmd { + args := []interface{}{"CF.RESERVE", key, capacity, "MAXITERATIONS", maxiterations} + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// CFReserveArgs creates an empty Cuckoo filter with the specified options. +// This function allows for specifying additional options such as bucket size and maximum number of iterations. +// For more information - https://redis.io/commands/cf.reserve/ +func (c cmdable) CFReserveArgs(ctx context.Context, key string, options *CFReserveOptions) *StatusCmd { + args := []interface{}{"CF.RESERVE", key, options.Capacity} + if options.BucketSize != 0 { + args = append(args, "BUCKETSIZE", options.BucketSize) + } + if options.MaxIterations != 0 { + args = append(args, "MAXITERATIONS", options.MaxIterations) + } + if options.Expansion != 0 { + args = append(args, "EXPANSION", options.Expansion) + } + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// CFAdd adds an element to a Cuckoo filter. +// Returns true if the element was added to the filter or false if it already exists in the filter. +// For more information - https://redis.io/commands/cf.add/ +func (c cmdable) CFAdd(ctx context.Context, key string, element interface{}) *BoolCmd { + args := []interface{}{"CF.ADD", key, element} + cmd := NewBoolCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// CFAddNX adds an element to a Cuckoo filter only if it does not already exist in the filter. +// Returns true if the element was added to the filter or false if it already exists in the filter. +// For more information - https://redis.io/commands/cf.addnx/ +func (c cmdable) CFAddNX(ctx context.Context, key string, element interface{}) *BoolCmd { + args := []interface{}{"CF.ADDNX", key, element} + cmd := NewBoolCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// CFCount returns an estimate of the number of times an element may be in a Cuckoo Filter. +// For more information - https://redis.io/commands/cf.count/ +func (c cmdable) CFCount(ctx context.Context, key string, element interface{}) *IntCmd { + args := []interface{}{"CF.COUNT", key, element} + cmd := NewIntCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// CFDel deletes an item once from the cuckoo filter. +// For more information - https://redis.io/commands/cf.del/ +func (c cmdable) CFDel(ctx context.Context, key string, element interface{}) *BoolCmd { + args := []interface{}{"CF.DEL", key, element} + cmd := NewBoolCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// CFExists determines whether an item may exist in the Cuckoo Filter or not. +// For more information - https://redis.io/commands/cf.exists/ +func (c cmdable) CFExists(ctx context.Context, key string, element interface{}) *BoolCmd { + args := []interface{}{"CF.EXISTS", key, element} + cmd := NewBoolCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// CFLoadChunk restores a filter previously saved using SCANDUMP. +// For more information - https://redis.io/commands/cf.loadchunk/ +func (c cmdable) CFLoadChunk(ctx context.Context, key string, iterator int64, data interface{}) *StatusCmd { + args := []interface{}{"CF.LOADCHUNK", key, iterator, data} + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// CFScanDump begins an incremental save of the cuckoo filter. +// For more information - https://redis.io/commands/cf.scandump/ +func (c cmdable) CFScanDump(ctx context.Context, key string, iterator int64) *ScanDumpCmd { + args := []interface{}{"CF.SCANDUMP", key, iterator} + cmd := newScanDumpCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +type CFInfo struct { + Size int64 + NumBuckets int64 + NumFilters int64 + NumItemsInserted int64 + NumItemsDeleted int64 + BucketSize int64 + ExpansionRate int64 + MaxIteration int64 +} + +type CFInfoCmd struct { + baseCmd + + val CFInfo +} + +func NewCFInfoCmd(ctx context.Context, args ...interface{}) *CFInfoCmd { + return &CFInfoCmd{ + baseCmd: baseCmd{ + ctx: ctx, + args: args, + }, + } +} + +func (cmd *CFInfoCmd) SetVal(val CFInfo) { + cmd.val = val +} + +func (cmd *CFInfoCmd) String() string { + return cmdString(cmd, cmd.val) +} + +func (cmd *CFInfoCmd) Val() CFInfo { + return cmd.val +} + +func (cmd *CFInfoCmd) Result() (CFInfo, error) { + return cmd.val, cmd.err +} + +func (cmd *CFInfoCmd) readReply(rd *proto.Reader) (err error) { + n, err := rd.ReadMapLen() + if err != nil { + return err + } + + var key string + var result CFInfo + for f := 0; f < n; f++ { + key, err = rd.ReadString() + if err != nil { + return err + } + + switch key { + case "Size": + result.Size, err = rd.ReadInt() + case "Number of buckets": + result.NumBuckets, err = rd.ReadInt() + case "Number of filters": + result.NumFilters, err = rd.ReadInt() + case "Number of items inserted": + result.NumItemsInserted, err = rd.ReadInt() + case "Number of items deleted": + result.NumItemsDeleted, err = rd.ReadInt() + case "Bucket size": + result.BucketSize, err = rd.ReadInt() + case "Expansion rate": + result.ExpansionRate, err = rd.ReadInt() + case "Max iterations": + result.MaxIteration, err = rd.ReadInt() + + default: + return fmt.Errorf("redis: CF.INFO unexpected key %s", key) + } + + if err != nil { + return err + } + } + + cmd.val = result + return nil +} + +// CFInfo returns information about a Cuckoo filter. +// For more information - https://redis.io/commands/cf.info/ +func (c cmdable) CFInfo(ctx context.Context, key string) *CFInfoCmd { + args := []interface{}{"CF.INFO", key} + cmd := NewCFInfoCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// CFInsert inserts elements into a Cuckoo filter. +// This function also allows for specifying additional options such as capacity, error rate, expansion rate, and non-scaling behavior. +// Returns an array of booleans indicating whether each element was added to the filter or not. +// For more information - https://redis.io/commands/cf.insert/ +func (c cmdable) CFInsert(ctx context.Context, key string, options *CFInsertOptions, elements ...interface{}) *BoolSliceCmd { + args := []interface{}{"CF.INSERT", key} + args = c.getCfInsertArgs(args, options, elements...) + + cmd := NewBoolSliceCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// CFInsertNX inserts elements into a Cuckoo filter only if they do not already exist in the filter. +// This function also allows for specifying additional options such as: +// capacity, error rate, expansion rate, and non-scaling behavior. +// Returns an array of integers indicating whether each element was added to the filter or not. +// For more information - https://redis.io/commands/cf.insertnx/ +func (c cmdable) CFInsertNX(ctx context.Context, key string, options *CFInsertOptions, elements ...interface{}) *IntSliceCmd { + args := []interface{}{"CF.INSERTNX", key} + args = c.getCfInsertArgs(args, options, elements...) + + cmd := NewIntSliceCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +func (c cmdable) getCfInsertArgs(args []interface{}, options *CFInsertOptions, elements ...interface{}) []interface{} { + if options != nil { + if options.Capacity != 0 { + args = append(args, "CAPACITY", options.Capacity) + } + if options.NoCreate { + args = append(args, "NOCREATE") + } + } + args = append(args, "ITEMS") + args = append(args, elements...) + + return args +} + +// CFMExists check if multiple elements exist in a Cuckoo filter. +// Returns an array of booleans indicating whether each element exists in the filter or not. +// For more information - https://redis.io/commands/cf.mexists/ +func (c cmdable) CFMExists(ctx context.Context, key string, elements ...interface{}) *BoolSliceCmd { + args := []interface{}{"CF.MEXISTS", key} + args = append(args, elements...) + cmd := NewBoolSliceCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// ------------------------------------------- +// CMS commands +//------------------------------------------- + +// CMSIncrBy increments the count of one or more items in a Count-Min Sketch filter. +// Returns an array of integers representing the updated count of each item. +// For more information - https://redis.io/commands/cms.incrby/ +func (c cmdable) CMSIncrBy(ctx context.Context, key string, elements ...interface{}) *IntSliceCmd { + args := make([]interface{}, 2, 2+len(elements)) + args[0] = "CMS.INCRBY" + args[1] = key + args = appendArgs(args, elements) + + cmd := NewIntSliceCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +type CMSInfo struct { + Width int64 + Depth int64 + Count int64 +} + +type CMSInfoCmd struct { + baseCmd + + val CMSInfo +} + +func NewCMSInfoCmd(ctx context.Context, args ...interface{}) *CMSInfoCmd { + return &CMSInfoCmd{ + baseCmd: baseCmd{ + ctx: ctx, + args: args, + }, + } +} + +func (cmd *CMSInfoCmd) SetVal(val CMSInfo) { + cmd.val = val +} + +func (cmd *CMSInfoCmd) String() string { + return cmdString(cmd, cmd.val) +} + +func (cmd *CMSInfoCmd) Val() CMSInfo { + return cmd.val +} + +func (cmd *CMSInfoCmd) Result() (CMSInfo, error) { + return cmd.val, cmd.err +} + +func (cmd *CMSInfoCmd) readReply(rd *proto.Reader) (err error) { + n, err := rd.ReadMapLen() + if err != nil { + return err + } + + var key string + var result CMSInfo + for f := 0; f < n; f++ { + key, err = rd.ReadString() + if err != nil { + return err + } + + switch key { + case "width": + result.Width, err = rd.ReadInt() + case "depth": + result.Depth, err = rd.ReadInt() + case "count": + result.Count, err = rd.ReadInt() + default: + return fmt.Errorf("redis: CMS.INFO unexpected key %s", key) + } + + if err != nil { + return err + } + } + + cmd.val = result + return nil +} + +// CMSInfo returns information about a Count-Min Sketch filter. +// For more information - https://redis.io/commands/cms.info/ +func (c cmdable) CMSInfo(ctx context.Context, key string) *CMSInfoCmd { + args := []interface{}{"CMS.INFO", key} + cmd := NewCMSInfoCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// CMSInitByDim creates an empty Count-Min Sketch filter with the specified dimensions. +// For more information - https://redis.io/commands/cms.initbydim/ +func (c cmdable) CMSInitByDim(ctx context.Context, key string, width, depth int64) *StatusCmd { + args := []interface{}{"CMS.INITBYDIM", key, width, depth} + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// CMSInitByProb creates an empty Count-Min Sketch filter with the specified error rate and probability. +// For more information - https://redis.io/commands/cms.initbyprob/ +func (c cmdable) CMSInitByProb(ctx context.Context, key string, errorRate, probability float64) *StatusCmd { + args := []interface{}{"CMS.INITBYPROB", key, errorRate, probability} + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// CMSMerge merges multiple Count-Min Sketch filters into a single filter. +// The destination filter must not exist and will be created with the dimensions of the first source filter. +// The number of items in each source filter must be equal. +// Returns OK on success or an error if the filters could not be merged. +// For more information - https://redis.io/commands/cms.merge/ +func (c cmdable) CMSMerge(ctx context.Context, destKey string, sourceKeys ...string) *StatusCmd { + args := []interface{}{"CMS.MERGE", destKey, len(sourceKeys)} + for _, s := range sourceKeys { + args = append(args, s) + } + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// CMSMergeWithWeight merges multiple Count-Min Sketch filters into a single filter with weights for each source filter. +// The destination filter must not exist and will be created with the dimensions of the first source filter. +// The number of items in each source filter must be equal. +// Returns OK on success or an error if the filters could not be merged. +// For more information - https://redis.io/commands/cms.merge/ +func (c cmdable) CMSMergeWithWeight(ctx context.Context, destKey string, sourceKeys map[string]int64) *StatusCmd { + args := make([]interface{}, 0, 4+(len(sourceKeys)*2+1)) + args = append(args, "CMS.MERGE", destKey, len(sourceKeys)) + + if len(sourceKeys) > 0 { + sk := make([]interface{}, len(sourceKeys)) + sw := make([]interface{}, len(sourceKeys)) + + i := 0 + for k, w := range sourceKeys { + sk[i] = k + sw[i] = w + i++ + } + + args = append(args, sk...) + args = append(args, "WEIGHTS") + args = append(args, sw...) + } + + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// CMSQuery returns count for item(s). +// For more information - https://redis.io/commands/cms.query/ +func (c cmdable) CMSQuery(ctx context.Context, key string, elements ...interface{}) *IntSliceCmd { + args := []interface{}{"CMS.QUERY", key} + args = append(args, elements...) + cmd := NewIntSliceCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// ------------------------------------------- +// TopK commands +//-------------------------------------------- + +// TopKAdd adds one or more elements to a Top-K filter. +// Returns an array of strings representing the items that were removed from the filter, if any. +// For more information - https://redis.io/commands/topk.add/ +func (c cmdable) TopKAdd(ctx context.Context, key string, elements ...interface{}) *StringSliceCmd { + args := make([]interface{}, 2, 2+len(elements)) + args[0] = "TOPK.ADD" + args[1] = key + args = appendArgs(args, elements) + + cmd := NewStringSliceCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// TopKReserve creates an empty Top-K filter with the specified number of top items to keep. +// For more information - https://redis.io/commands/topk.reserve/ +func (c cmdable) TopKReserve(ctx context.Context, key string, k int64) *StatusCmd { + args := []interface{}{"TOPK.RESERVE", key, k} + + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// TopKReserveWithOptions creates an empty Top-K filter with the specified number of top items to keep and additional options. +// This function allows for specifying additional options such as width, depth and decay. +// For more information - https://redis.io/commands/topk.reserve/ +func (c cmdable) TopKReserveWithOptions(ctx context.Context, key string, k int64, width, depth int64, decay float64) *StatusCmd { + args := []interface{}{"TOPK.RESERVE", key, k, width, depth, decay} + + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +type TopKInfo struct { + K int64 + Width int64 + Depth int64 + Decay float64 +} + +type TopKInfoCmd struct { + baseCmd + + val TopKInfo +} + +func NewTopKInfoCmd(ctx context.Context, args ...interface{}) *TopKInfoCmd { + return &TopKInfoCmd{ + baseCmd: baseCmd{ + ctx: ctx, + args: args, + }, + } +} + +func (cmd *TopKInfoCmd) SetVal(val TopKInfo) { + cmd.val = val +} + +func (cmd *TopKInfoCmd) String() string { + return cmdString(cmd, cmd.val) +} + +func (cmd *TopKInfoCmd) Val() TopKInfo { + return cmd.val +} + +func (cmd *TopKInfoCmd) Result() (TopKInfo, error) { + return cmd.val, cmd.err +} + +func (cmd *TopKInfoCmd) readReply(rd *proto.Reader) (err error) { + n, err := rd.ReadMapLen() + if err != nil { + return err + } + + var key string + var result TopKInfo + for f := 0; f < n; f++ { + key, err = rd.ReadString() + if err != nil { + return err + } + + switch key { + case "k": + result.K, err = rd.ReadInt() + case "width": + result.Width, err = rd.ReadInt() + case "depth": + result.Depth, err = rd.ReadInt() + case "decay": + result.Decay, err = rd.ReadFloat() + default: + return fmt.Errorf("redis: topk.info unexpected key %s", key) + } + + if err != nil { + return err + } + } + + cmd.val = result + return nil +} + +// TopKInfo returns information about a Top-K filter. +// For more information - https://redis.io/commands/topk.info/ +func (c cmdable) TopKInfo(ctx context.Context, key string) *TopKInfoCmd { + args := []interface{}{"TOPK.INFO", key} + + cmd := NewTopKInfoCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// TopKQuery check if multiple elements exist in a Top-K filter. +// Returns an array of booleans indicating whether each element exists in the filter or not. +// For more information - https://redis.io/commands/topk.query/ +func (c cmdable) TopKQuery(ctx context.Context, key string, elements ...interface{}) *BoolSliceCmd { + args := make([]interface{}, 2, 2+len(elements)) + args[0] = "TOPK.QUERY" + args[1] = key + args = appendArgs(args, elements) + + cmd := NewBoolSliceCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// TopKCount returns an estimate of the number of times an item may be in a Top-K filter. +// For more information - https://redis.io/commands/topk.count/ +func (c cmdable) TopKCount(ctx context.Context, key string, elements ...interface{}) *IntSliceCmd { + args := make([]interface{}, 2, 2+len(elements)) + args[0] = "TOPK.COUNT" + args[1] = key + args = appendArgs(args, elements) + + cmd := NewIntSliceCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// TopKIncrBy increases the count of one or more items in a Top-K filter. +// For more information - https://redis.io/commands/topk.incrby/ +func (c cmdable) TopKIncrBy(ctx context.Context, key string, elements ...interface{}) *StringSliceCmd { + args := make([]interface{}, 2, 2+len(elements)) + args[0] = "TOPK.INCRBY" + args[1] = key + args = appendArgs(args, elements) + + cmd := NewStringSliceCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// TopKList returns all items in Top-K list. +// For more information - https://redis.io/commands/topk.list/ +func (c cmdable) TopKList(ctx context.Context, key string) *StringSliceCmd { + args := []interface{}{"TOPK.LIST", key} + + cmd := NewStringSliceCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// TopKListWithCount returns all items in Top-K list with their respective count. +// For more information - https://redis.io/commands/topk.list/ +func (c cmdable) TopKListWithCount(ctx context.Context, key string) *MapStringIntCmd { + args := []interface{}{"TOPK.LIST", key, "WITHCOUNT"} + + cmd := NewMapStringIntCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// ------------------------------------------- +// t-digest commands +// -------------------------------------------- + +// TDigestAdd adds one or more elements to a t-Digest data structure. +// Returns OK on success or an error if the operation could not be completed. +// For more information - https://redis.io/commands/tdigest.add/ +func (c cmdable) TDigestAdd(ctx context.Context, key string, elements ...float64) *StatusCmd { + args := make([]interface{}, 2, 2+len(elements)) + args[0] = "TDIGEST.ADD" + args[1] = key + + // Convert floatSlice to []interface{} + interfaceSlice := make([]interface{}, len(elements)) + for i, v := range elements { + interfaceSlice[i] = v + } + + args = append(args, interfaceSlice...) + + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// TDigestByRank returns an array of values from a t-Digest data structure based on their rank. +// The rank of an element is its position in the sorted list of all elements in the t-Digest. +// Returns an array of floats representing the values at the specified ranks or an error if the operation could not be completed. +// For more information - https://redis.io/commands/tdigest.byrank/ +func (c cmdable) TDigestByRank(ctx context.Context, key string, rank ...uint64) *FloatSliceCmd { + args := make([]interface{}, 2, 2+len(rank)) + args[0] = "TDIGEST.BYRANK" + args[1] = key + + // Convert uint slice to []interface{} + interfaceSlice := make([]interface{}, len(rank)) + for i, v := range rank { + interfaceSlice[i] = v + } + + args = append(args, interfaceSlice...) + + cmd := NewFloatSliceCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// TDigestByRevRank returns an array of values from a t-Digest data structure based on their reverse rank. +// The reverse rank of an element is its position in the sorted list of all elements in the t-Digest when sorted in descending order. +// Returns an array of floats representing the values at the specified ranks or an error if the operation could not be completed. +// For more information - https://redis.io/commands/tdigest.byrevrank/ +func (c cmdable) TDigestByRevRank(ctx context.Context, key string, rank ...uint64) *FloatSliceCmd { + args := make([]interface{}, 2, 2+len(rank)) + args[0] = "TDIGEST.BYREVRANK" + args[1] = key + + // Convert uint slice to []interface{} + interfaceSlice := make([]interface{}, len(rank)) + for i, v := range rank { + interfaceSlice[i] = v + } + + args = append(args, interfaceSlice...) + + cmd := NewFloatSliceCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// TDigestCDF returns an array of cumulative distribution function (CDF) values for one or more elements in a t-Digest data structure. +// The CDF value for an element is the fraction of all elements in the t-Digest that are less than or equal to it. +// Returns an array of floats representing the CDF values for each element or an error if the operation could not be completed. +// For more information - https://redis.io/commands/tdigest.cdf/ +func (c cmdable) TDigestCDF(ctx context.Context, key string, elements ...float64) *FloatSliceCmd { + args := make([]interface{}, 2, 2+len(elements)) + args[0] = "TDIGEST.CDF" + args[1] = key + + // Convert floatSlice to []interface{} + interfaceSlice := make([]interface{}, len(elements)) + for i, v := range elements { + interfaceSlice[i] = v + } + + args = append(args, interfaceSlice...) + + cmd := NewFloatSliceCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// TDigestCreate creates an empty t-Digest data structure with default parameters. +// Returns OK on success or an error if the operation could not be completed. +// For more information - https://redis.io/commands/tdigest.create/ +func (c cmdable) TDigestCreate(ctx context.Context, key string) *StatusCmd { + args := []interface{}{"TDIGEST.CREATE", key} + + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// TDigestCreateWithCompression creates an empty t-Digest data structure with a specified compression parameter. +// The compression parameter controls the accuracy and memory usage of the t-Digest. +// Returns OK on success or an error if the operation could not be completed. +// For more information - https://redis.io/commands/tdigest.create/ +func (c cmdable) TDigestCreateWithCompression(ctx context.Context, key string, compression int64) *StatusCmd { + args := []interface{}{"TDIGEST.CREATE", key, "COMPRESSION", compression} + + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +type TDigestInfo struct { + Compression int64 + Capacity int64 + MergedNodes int64 + UnmergedNodes int64 + MergedWeight int64 + UnmergedWeight int64 + Observations int64 + TotalCompressions int64 + MemoryUsage int64 +} + +type TDigestInfoCmd struct { + baseCmd + + val TDigestInfo +} + +func NewTDigestInfoCmd(ctx context.Context, args ...interface{}) *TDigestInfoCmd { + return &TDigestInfoCmd{ + baseCmd: baseCmd{ + ctx: ctx, + args: args, + }, + } +} + +func (cmd *TDigestInfoCmd) SetVal(val TDigestInfo) { + cmd.val = val +} + +func (cmd *TDigestInfoCmd) String() string { + return cmdString(cmd, cmd.val) +} + +func (cmd *TDigestInfoCmd) Val() TDigestInfo { + return cmd.val +} + +func (cmd *TDigestInfoCmd) Result() (TDigestInfo, error) { + return cmd.val, cmd.err +} + +func (cmd *TDigestInfoCmd) readReply(rd *proto.Reader) (err error) { + n, err := rd.ReadMapLen() + if err != nil { + return err + } + + var key string + var result TDigestInfo + for f := 0; f < n; f++ { + key, err = rd.ReadString() + if err != nil { + return err + } + + switch key { + case "Compression": + result.Compression, err = rd.ReadInt() + case "Capacity": + result.Capacity, err = rd.ReadInt() + case "Merged nodes": + result.MergedNodes, err = rd.ReadInt() + case "Unmerged nodes": + result.UnmergedNodes, err = rd.ReadInt() + case "Merged weight": + result.MergedWeight, err = rd.ReadInt() + case "Unmerged weight": + result.UnmergedWeight, err = rd.ReadInt() + case "Observations": + result.Observations, err = rd.ReadInt() + case "Total compressions": + result.TotalCompressions, err = rd.ReadInt() + case "Memory usage": + result.MemoryUsage, err = rd.ReadInt() + default: + return fmt.Errorf("redis: tdigest.info unexpected key %s", key) + } + + if err != nil { + return err + } + } + + cmd.val = result + return nil +} + +// TDigestInfo returns information about a t-Digest data structure. +// For more information - https://redis.io/commands/tdigest.info/ +func (c cmdable) TDigestInfo(ctx context.Context, key string) *TDigestInfoCmd { + args := []interface{}{"TDIGEST.INFO", key} + + cmd := NewTDigestInfoCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// TDigestMax returns the maximum value from a t-Digest data structure. +// For more information - https://redis.io/commands/tdigest.max/ +func (c cmdable) TDigestMax(ctx context.Context, key string) *FloatCmd { + args := []interface{}{"TDIGEST.MAX", key} + + cmd := NewFloatCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +type TDigestMergeOptions struct { + Compression int64 + Override bool +} + +// TDigestMerge merges multiple t-Digest data structures into a single t-Digest. +// This function also allows for specifying additional options such as compression and override behavior. +// Returns OK on success or an error if the operation could not be completed. +// For more information - https://redis.io/commands/tdigest.merge/ +func (c cmdable) TDigestMerge(ctx context.Context, destKey string, options *TDigestMergeOptions, sourceKeys ...string) *StatusCmd { + args := []interface{}{"TDIGEST.MERGE", destKey, len(sourceKeys)} + + for _, sourceKey := range sourceKeys { + args = append(args, sourceKey) + } + + if options != nil { + if options.Compression != 0 { + args = append(args, "COMPRESSION", options.Compression) + } + if options.Override { + args = append(args, "OVERRIDE") + } + } + + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// TDigestMin returns the minimum value from a t-Digest data structure. +// For more information - https://redis.io/commands/tdigest.min/ +func (c cmdable) TDigestMin(ctx context.Context, key string) *FloatCmd { + args := []interface{}{"TDIGEST.MIN", key} + + cmd := NewFloatCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// TDigestQuantile returns an array of quantile values for one or more elements in a t-Digest data structure. +// The quantile value for an element is the fraction of all elements in the t-Digest that are less than or equal to it. +// Returns an array of floats representing the quantile values for each element or an error if the operation could not be completed. +// For more information - https://redis.io/commands/tdigest.quantile/ +func (c cmdable) TDigestQuantile(ctx context.Context, key string, elements ...float64) *FloatSliceCmd { + args := make([]interface{}, 2, 2+len(elements)) + args[0] = "TDIGEST.QUANTILE" + args[1] = key + + // Convert floatSlice to []interface{} + interfaceSlice := make([]interface{}, len(elements)) + for i, v := range elements { + interfaceSlice[i] = v + } + + args = append(args, interfaceSlice...) + + cmd := NewFloatSliceCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// TDigestRank returns an array of rank values for one or more elements in a t-Digest data structure. +// The rank of an element is its position in the sorted list of all elements in the t-Digest. +// Returns an array of integers representing the rank values for each element or an error if the operation could not be completed. +// For more information - https://redis.io/commands/tdigest.rank/ +func (c cmdable) TDigestRank(ctx context.Context, key string, values ...float64) *IntSliceCmd { + args := make([]interface{}, 2, 2+len(values)) + args[0] = "TDIGEST.RANK" + args[1] = key + + // Convert floatSlice to []interface{} + interfaceSlice := make([]interface{}, len(values)) + for i, v := range values { + interfaceSlice[i] = v + } + + args = append(args, interfaceSlice...) + + cmd := NewIntSliceCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// TDigestReset resets a t-Digest data structure to its initial state. +// Returns OK on success or an error if the operation could not be completed. +// For more information - https://redis.io/commands/tdigest.reset/ +func (c cmdable) TDigestReset(ctx context.Context, key string) *StatusCmd { + args := []interface{}{"TDIGEST.RESET", key} + + cmd := NewStatusCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// TDigestRevRank returns an array of reverse rank values for one or more elements in a t-Digest data structure. +// The reverse rank of an element is its position in the sorted list of all elements in the t-Digest when sorted in descending order. +// Returns an array of integers representing the reverse rank values for each element or an error if the operation could not be completed. +// For more information - https://redis.io/commands/tdigest.revrank/ +func (c cmdable) TDigestRevRank(ctx context.Context, key string, values ...float64) *IntSliceCmd { + args := make([]interface{}, 2, 2+len(values)) + args[0] = "TDIGEST.REVRANK" + args[1] = key + + // Convert floatSlice to []interface{} + interfaceSlice := make([]interface{}, len(values)) + for i, v := range values { + interfaceSlice[i] = v + } + + args = append(args, interfaceSlice...) + + cmd := NewIntSliceCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} + +// TDigestTrimmedMean returns the trimmed mean value from a t-Digest data structure. +// The trimmed mean is calculated by removing a specified fraction of the highest and lowest values from the t-Digest and then calculating the mean of the remaining values. +// Returns a float representing the trimmed mean value or an error if the operation could not be completed. +// For more information - https://redis.io/commands/tdigest.trimmed_mean/ +func (c cmdable) TDigestTrimmedMean(ctx context.Context, key string, lowCutQuantile, highCutQuantile float64) *FloatCmd { + args := []interface{}{"TDIGEST.TRIMMED_MEAN", key, lowCutQuantile, highCutQuantile} + + cmd := NewFloatCmd(ctx, args...) + _ = c(ctx, cmd) + return cmd +} diff --git a/probabilistic_test.go b/probabilistic_test.go new file mode 100644 index 00000000..4b0dde64 --- /dev/null +++ b/probabilistic_test.go @@ -0,0 +1,739 @@ +package redis_test + +import ( + "context" + "fmt" + "math" + + . "github.com/bsm/ginkgo/v2" + . "github.com/bsm/gomega" + "github.com/redis/go-redis/v9" +) + +var _ = Describe("Probabilistic commands", Label("probabilistic"), func() { + ctx := context.TODO() + var client *redis.Client + + BeforeEach(func() { + client = redis.NewClient(&redis.Options{Addr: ":6379"}) + Expect(client.FlushDB(ctx).Err()).NotTo(HaveOccurred()) + }) + + AfterEach(func() { + Expect(client.Close()).NotTo(HaveOccurred()) + }) + + Describe("bloom", Label("bloom"), func() { + It("should BFAdd", Label("bloom", "bfadd"), func() { + resultAdd, err := client.BFAdd(ctx, "testbf1", 1).Result() + + Expect(err).NotTo(HaveOccurred()) + Expect(resultAdd).To(BeTrue()) + + resultInfo, err := client.BFInfo(ctx, "testbf1").Result() + + Expect(err).NotTo(HaveOccurred()) + Expect(resultInfo).To(BeAssignableToTypeOf(redis.BFInfo{})) + Expect(resultInfo.ItemsInserted).To(BeEquivalentTo(int64(1))) + }) + + It("should BFCard", Label("bloom", "bfcard"), func() { + // This is a probabilistic data structure, and it's not always guaranteed that we will get back + // the exact number of inserted items, during hash collisions + // But with such a low number of items (only 3), + // the probability of a collision is very low, so we can expect to get back the exact number of items + _, err := client.BFAdd(ctx, "testbf1", "item1").Result() + Expect(err).NotTo(HaveOccurred()) + _, err = client.BFAdd(ctx, "testbf1", "item2").Result() + Expect(err).NotTo(HaveOccurred()) + _, err = client.BFAdd(ctx, "testbf1", 3).Result() + Expect(err).NotTo(HaveOccurred()) + + result, err := client.BFCard(ctx, "testbf1").Result() + + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(BeEquivalentTo(int64(3))) + }) + + It("should BFExists", Label("bloom", "bfexists"), func() { + exists, err := client.BFExists(ctx, "testbf1", "item1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(exists).To(BeFalse()) + + _, err = client.BFAdd(ctx, "testbf1", "item1").Result() + Expect(err).NotTo(HaveOccurred()) + + exists, err = client.BFExists(ctx, "testbf1", "item1").Result() + + Expect(err).NotTo(HaveOccurred()) + Expect(exists).To(BeTrue()) + }) + + It("should BFInfo and BFReserve", Label("bloom", "bfinfo", "bfreserve"), func() { + err := client.BFReserve(ctx, "testbf1", 0.001, 2000).Err() + Expect(err).NotTo(HaveOccurred()) + + result, err := client.BFInfo(ctx, "testbf1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(BeAssignableToTypeOf(redis.BFInfo{})) + Expect(result.Capacity).To(BeEquivalentTo(int64(2000))) + }) + + It("should BFInfoCapacity, BFInfoSize, BFInfoFilters, BFInfoItems, BFInfoExpansion, ", Label("bloom", "bfinfocapacity", "bfinfosize", "bfinfofilters", "bfinfoitems", "bfinfoexpansion"), func() { + err := client.BFReserve(ctx, "testbf1", 0.001, 2000).Err() + Expect(err).NotTo(HaveOccurred()) + + result, err := client.BFInfoCapacity(ctx, "testbf1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(result.Capacity).To(BeEquivalentTo(int64(2000))) + + result, err = client.BFInfoItems(ctx, "testbf1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(result.ItemsInserted).To(BeEquivalentTo(int64(0))) + + result, err = client.BFInfoSize(ctx, "testbf1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(result.Size).To(BeEquivalentTo(int64(4056))) + + err = client.BFReserveExpansion(ctx, "testbf2", 0.001, 2000, 3).Err() + Expect(err).NotTo(HaveOccurred()) + + result, err = client.BFInfoFilters(ctx, "testbf2").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(result.Filters).To(BeEquivalentTo(int64(1))) + + result, err = client.BFInfoExpansion(ctx, "testbf2").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(result.ExpansionRate).To(BeEquivalentTo(int64(3))) + }) + + It("should BFInsert", Label("bloom", "bfinsert"), func() { + options := &redis.BFInsertOptions{ + Capacity: 2000, + Error: 0.001, + Expansion: 3, + NonScaling: false, + NoCreate: true, + } + + resultInsert, err := client.BFInsert(ctx, "testbf1", options, "item1").Result() + Expect(err).To(HaveOccurred()) + Expect(err).To(MatchError("ERR not found")) + + options = &redis.BFInsertOptions{ + Capacity: 2000, + Error: 0.001, + Expansion: 3, + NonScaling: false, + NoCreate: false, + } + + resultInsert, err = client.BFInsert(ctx, "testbf1", options, "item1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(resultInsert)).To(BeEquivalentTo(1)) + + exists, err := client.BFExists(ctx, "testbf1", "item1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(exists).To(BeTrue()) + + result, err := client.BFInfo(ctx, "testbf1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(BeAssignableToTypeOf(redis.BFInfo{})) + Expect(result.Capacity).To(BeEquivalentTo(int64(2000))) + Expect(result.ExpansionRate).To(BeEquivalentTo(int64(3))) + }) + + It("should BFMAdd", Label("bloom", "bfmadd"), func() { + resultAdd, err := client.BFMAdd(ctx, "testbf1", "item1", "item2", "item3").Result() + + Expect(err).NotTo(HaveOccurred()) + Expect(len(resultAdd)).To(Equal(3)) + + resultInfo, err := client.BFInfo(ctx, "testbf1").Result() + + Expect(err).NotTo(HaveOccurred()) + Expect(resultInfo).To(BeAssignableToTypeOf(redis.BFInfo{})) + Expect(resultInfo.ItemsInserted).To(BeEquivalentTo(int64(3))) + resultAdd2, err := client.BFMAdd(ctx, "testbf1", "item1", "item2", "item4").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(resultAdd2[0]).To(BeFalse()) + Expect(resultAdd2[1]).To(BeFalse()) + Expect(resultAdd2[2]).To(BeTrue()) + + }) + + It("should BFMExists", Label("bloom", "bfmexists"), func() { + exist, err := client.BFMExists(ctx, "testbf1", "item1", "item2", "item3").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(exist)).To(Equal(3)) + Expect(exist[0]).To(BeFalse()) + Expect(exist[1]).To(BeFalse()) + Expect(exist[2]).To(BeFalse()) + + _, err = client.BFMAdd(ctx, "testbf1", "item1", "item2", "item3").Result() + Expect(err).NotTo(HaveOccurred()) + + exist, err = client.BFMExists(ctx, "testbf1", "item1", "item2", "item3", "item4").Result() + + Expect(err).NotTo(HaveOccurred()) + Expect(len(exist)).To(Equal(4)) + Expect(exist[0]).To(BeTrue()) + Expect(exist[1]).To(BeTrue()) + Expect(exist[2]).To(BeTrue()) + Expect(exist[3]).To(BeFalse()) + }) + + It("should BFReserveExpansion", Label("bloom", "bfreserveexpansion"), func() { + err := client.BFReserveExpansion(ctx, "testbf1", 0.001, 2000, 3).Err() + Expect(err).NotTo(HaveOccurred()) + + result, err := client.BFInfo(ctx, "testbf1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(BeAssignableToTypeOf(redis.BFInfo{})) + Expect(result.Capacity).To(BeEquivalentTo(int64(2000))) + Expect(result.ExpansionRate).To(BeEquivalentTo(int64(3))) + }) + + It("should BFReserveNonScaling", Label("bloom", "bfreservenonscaling"), func() { + err := client.BFReserveNonScaling(ctx, "testbfns1", 0.001, 1000).Err() + Expect(err).NotTo(HaveOccurred()) + + _, err = client.BFInfo(ctx, "testbfns1").Result() + Expect(err).To(HaveOccurred()) + }) + + It("should BFScanDump and BFLoadChunk", Label("bloom", "bfscandump", "bfloadchunk"), func() { + err := client.BFReserve(ctx, "testbfsd1", 0.001, 3000).Err() + Expect(err).NotTo(HaveOccurred()) + for i := 0; i < 1000; i++ { + client.BFAdd(ctx, "testbfsd1", i) + } + infBefore := client.BFInfoSize(ctx, "testbfsd1") + fd := []redis.ScanDump{} + sd, err := client.BFScanDump(ctx, "testbfsd1", 0).Result() + for { + if sd.Iter == 0 { + break + } + Expect(err).NotTo(HaveOccurred()) + fd = append(fd, sd) + sd, err = client.BFScanDump(ctx, "testbfsd1", sd.Iter).Result() + } + client.Del(ctx, "testbfsd1") + for _, e := range fd { + client.BFLoadChunk(ctx, "testbfsd1", e.Iter, e.Data) + } + infAfter := client.BFInfoSize(ctx, "testbfsd1") + Expect(infBefore).To(BeEquivalentTo(infAfter)) + }) + + It("should BFReserveArgs", Label("bloom", "bfreserveargs"), func() { + options := &redis.BFReserveOptions{ + Capacity: 2000, + Error: 0.001, + Expansion: 3, + NonScaling: false, + } + err := client.BFReserveArgs(ctx, "testbf", options).Err() + Expect(err).NotTo(HaveOccurred()) + + result, err := client.BFInfo(ctx, "testbf").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(BeAssignableToTypeOf(redis.BFInfo{})) + Expect(result.Capacity).To(BeEquivalentTo(int64(2000))) + Expect(result.ExpansionRate).To(BeEquivalentTo(int64(3))) + }) + }) + + Describe("cuckoo", Label("cuckoo"), func() { + It("should CFAdd", Label("cuckoo", "cfadd"), func() { + add, err := client.CFAdd(ctx, "testcf1", "item1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(add).To(BeTrue()) + + exists, err := client.CFExists(ctx, "testcf1", "item1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(exists).To(BeTrue()) + + info, err := client.CFInfo(ctx, "testcf1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(info).To(BeAssignableToTypeOf(redis.CFInfo{})) + Expect(info.NumItemsInserted).To(BeEquivalentTo(int64(1))) + }) + + It("should CFAddNX", Label("cuckoo", "cfaddnx"), func() { + add, err := client.CFAddNX(ctx, "testcf1", "item1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(add).To(BeTrue()) + + exists, err := client.CFExists(ctx, "testcf1", "item1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(exists).To(BeTrue()) + + result, err := client.CFAddNX(ctx, "testcf1", "item1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(BeFalse()) + + info, err := client.CFInfo(ctx, "testcf1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(info).To(BeAssignableToTypeOf(redis.CFInfo{})) + Expect(info.NumItemsInserted).To(BeEquivalentTo(int64(1))) + }) + + It("should CFCount", Label("cuckoo", "cfcount"), func() { + err := client.CFAdd(ctx, "testcf1", "item1").Err() + cnt, err := client.CFCount(ctx, "testcf1", "item1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(cnt).To(BeEquivalentTo(int64(1))) + + err = client.CFAdd(ctx, "testcf1", "item1").Err() + Expect(err).NotTo(HaveOccurred()) + + cnt, err = client.CFCount(ctx, "testcf1", "item1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(cnt).To(BeEquivalentTo(int64(2))) + }) + + It("should CFDel and CFExists", Label("cuckoo", "cfdel", "cfexists"), func() { + err := client.CFAdd(ctx, "testcf1", "item1").Err() + Expect(err).NotTo(HaveOccurred()) + + exists, err := client.CFExists(ctx, "testcf1", "item1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(exists).To(BeTrue()) + + del, err := client.CFDel(ctx, "testcf1", "item1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(del).To(BeTrue()) + + exists, err = client.CFExists(ctx, "testcf1", "item1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(exists).To(BeFalse()) + }) + + It("should CFInfo and CFReserve", Label("cuckoo", "cfinfo", "cfreserve"), func() { + err := client.CFReserve(ctx, "testcf1", 1000).Err() + Expect(err).NotTo(HaveOccurred()) + err = client.CFReserveExpansion(ctx, "testcfe1", 1000, 1).Err() + Expect(err).NotTo(HaveOccurred()) + err = client.CFReserveBucketSize(ctx, "testcfbs1", 1000, 4).Err() + Expect(err).NotTo(HaveOccurred()) + err = client.CFReserveMaxIterations(ctx, "testcfmi1", 1000, 10).Err() + Expect(err).NotTo(HaveOccurred()) + + result, err := client.CFInfo(ctx, "testcf1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(BeAssignableToTypeOf(redis.CFInfo{})) + }) + + It("should CFScanDump and CFLoadChunk", Label("bloom", "cfscandump", "cfloadchunk"), func() { + err := client.CFReserve(ctx, "testcfsd1", 1000).Err() + Expect(err).NotTo(HaveOccurred()) + for i := 0; i < 1000; i++ { + Item := fmt.Sprintf("item%d", i) + client.CFAdd(ctx, "testcfsd1", Item) + } + infBefore := client.CFInfo(ctx, "testcfsd1") + fd := []redis.ScanDump{} + sd, err := client.CFScanDump(ctx, "testcfsd1", 0).Result() + for { + if sd.Iter == 0 { + break + } + Expect(err).NotTo(HaveOccurred()) + fd = append(fd, sd) + sd, err = client.CFScanDump(ctx, "testcfsd1", sd.Iter).Result() + } + client.Del(ctx, "testcfsd1") + for _, e := range fd { + client.CFLoadChunk(ctx, "testcfsd1", e.Iter, e.Data) + } + infAfter := client.CFInfo(ctx, "testcfsd1") + Expect(infBefore).To(BeEquivalentTo(infAfter)) + }) + + It("should CFInfo and CFReserveArgs", Label("cuckoo", "cfinfo", "cfreserveargs"), func() { + args := &redis.CFReserveOptions{ + Capacity: 2048, + BucketSize: 3, + MaxIterations: 15, + Expansion: 2, + } + + err := client.CFReserveArgs(ctx, "testcf1", args).Err() + Expect(err).NotTo(HaveOccurred()) + + result, err := client.CFInfo(ctx, "testcf1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(BeAssignableToTypeOf(redis.CFInfo{})) + Expect(result.BucketSize).To(BeEquivalentTo(int64(3))) + Expect(result.MaxIteration).To(BeEquivalentTo(int64(15))) + Expect(result.ExpansionRate).To(BeEquivalentTo(int64(2))) + }) + + It("should CFInsert", Label("cuckoo", "cfinsert"), func() { + args := &redis.CFInsertOptions{ + Capacity: 3000, + NoCreate: true, + } + + result, err := client.CFInsert(ctx, "testcf1", args, "item1", "item2", "item3").Result() + Expect(err).To(HaveOccurred()) + + args = &redis.CFInsertOptions{ + Capacity: 3000, + NoCreate: false, + } + + result, err = client.CFInsert(ctx, "testcf1", args, "item1", "item2", "item3").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(result)).To(BeEquivalentTo(3)) + }) + + It("should CFInsertNX", Label("cuckoo", "cfinsertnx"), func() { + args := &redis.CFInsertOptions{ + Capacity: 3000, + NoCreate: true, + } + + result, err := client.CFInsertNX(ctx, "testcf1", args, "item1", "item2", "item2").Result() + Expect(err).To(HaveOccurred()) + + args = &redis.CFInsertOptions{ + Capacity: 3000, + NoCreate: false, + } + + 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))) + Expect(result[1]).To(BeEquivalentTo(int64(1))) + Expect(result[2]).To(BeEquivalentTo(int64(0))) + }) + + It("should CFMexists", Label("cuckoo", "cfmexists"), func() { + err := client.CFInsert(ctx, "testcf1", nil, "item1", "item2", "item3").Err() + Expect(err).NotTo(HaveOccurred()) + + result, err := client.CFMExists(ctx, "testcf1", "item1", "item2", "item3", "item4").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(result)).To(BeEquivalentTo(4)) + Expect(result[0]).To(BeTrue()) + Expect(result[1]).To(BeTrue()) + Expect(result[2]).To(BeTrue()) + Expect(result[3]).To(BeFalse()) + }) + + }) + + Describe("CMS", Label("cms"), func() { + It("should CMSIncrBy", Label("cms", "cmsincrby"), func() { + err := client.CMSInitByDim(ctx, "testcms1", 5, 10).Err() + Expect(err).NotTo(HaveOccurred()) + + result, err := client.CMSIncrBy(ctx, "testcms1", "item1", 1, "item2", 2, "item3", 3).Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(result)).To(BeEquivalentTo(3)) + Expect(result[0]).To(BeEquivalentTo(int64(1))) + Expect(result[1]).To(BeEquivalentTo(int64(2))) + Expect(result[2]).To(BeEquivalentTo(int64(3))) + + }) + + It("should CMSInitByDim and CMSInfo", Label("cms", "cmsinitbydim", "cmsinfo"), func() { + err := client.CMSInitByDim(ctx, "testcms1", 5, 10).Err() + Expect(err).NotTo(HaveOccurred()) + + info, err := client.CMSInfo(ctx, "testcms1").Result() + Expect(err).NotTo(HaveOccurred()) + + Expect(info).To(BeAssignableToTypeOf(redis.CMSInfo{})) + Expect(info.Width).To(BeEquivalentTo(int64(5))) + Expect(info.Depth).To(BeEquivalentTo(int64(10))) + }) + + It("should CMSInitByProb", Label("cms", "cmsinitbyprob"), func() { + err := client.CMSInitByProb(ctx, "testcms1", 0.002, 0.01).Err() + Expect(err).NotTo(HaveOccurred()) + + info, err := client.CMSInfo(ctx, "testcms1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(info).To(BeAssignableToTypeOf(redis.CMSInfo{})) + }) + + It("should CMSMerge, CMSMergeWithWeight and CMSQuery", Label("cms", "cmsmerge", "cmsquery"), func() { + err := client.CMSMerge(ctx, "destCms1", "testcms2", "testcms3").Err() + Expect(err).To(HaveOccurred()) + Expect(err).To(MatchError("CMS: key does not exist")) + + err = client.CMSInitByDim(ctx, "destCms1", 5, 10).Err() + Expect(err).NotTo(HaveOccurred()) + err = client.CMSInitByDim(ctx, "destCms2", 5, 10).Err() + Expect(err).NotTo(HaveOccurred()) + err = client.CMSInitByDim(ctx, "cms1", 2, 20).Err() + Expect(err).NotTo(HaveOccurred()) + err = client.CMSInitByDim(ctx, "cms2", 3, 20).Err() + Expect(err).NotTo(HaveOccurred()) + + err = client.CMSMerge(ctx, "destCms1", "cms1", "cms2").Err() + Expect(err).To(MatchError("CMS: width/depth is not equal")) + + client.Del(ctx, "cms1", "cms2") + + err = client.CMSInitByDim(ctx, "cms1", 5, 10).Err() + Expect(err).NotTo(HaveOccurred()) + err = client.CMSInitByDim(ctx, "cms2", 5, 10).Err() + Expect(err).NotTo(HaveOccurred()) + + client.CMSIncrBy(ctx, "cms1", "item1", 1, "item2", 2) + client.CMSIncrBy(ctx, "cms2", "item2", 2, "item3", 3) + + err = client.CMSMerge(ctx, "destCms1", "cms1", "cms2").Err() + Expect(err).NotTo(HaveOccurred()) + + result, err := client.CMSQuery(ctx, "destCms1", "item1", "item2", "item3").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(result)).To(BeEquivalentTo(3)) + Expect(result[0]).To(BeEquivalentTo(int64(1))) + Expect(result[1]).To(BeEquivalentTo(int64(4))) + Expect(result[2]).To(BeEquivalentTo(int64(3))) + + sourceSketches := map[string]int64{ + "cms1": 1, + "cms2": 2, + } + err = client.CMSMergeWithWeight(ctx, "destCms2", sourceSketches).Err() + Expect(err).NotTo(HaveOccurred()) + + result, err = client.CMSQuery(ctx, "destCms2", "item1", "item2", "item3").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(result)).To(BeEquivalentTo(3)) + Expect(result[0]).To(BeEquivalentTo(int64(1))) + Expect(result[1]).To(BeEquivalentTo(int64(6))) + Expect(result[2]).To(BeEquivalentTo(int64(6))) + + }) + + }) + + Describe("TopK", Label("topk"), func() { + It("should TopKReserve, TopKInfo, TopKAdd, TopKQuery, TopKCount, TopKIncrBy, TopKList, TopKListWithCount", Label("topk", "topkreserve", "topkinfo", "topkadd", "topkquery", "topkcount", "topkincrby", "topklist", "topklistwithcount"), func() { + err := client.TopKReserve(ctx, "topk1", 3).Err() + Expect(err).NotTo(HaveOccurred()) + + resultInfo, err := client.TopKInfo(ctx, "topk1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(resultInfo.K).To(BeEquivalentTo(int64(3))) + + resultAdd, err := client.TopKAdd(ctx, "topk1", "item1", "item2", 3, "item1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(resultAdd)).To(BeEquivalentTo(int64(4))) + + resultQuery, err := client.TopKQuery(ctx, "topk1", "item1", "item2", 4, 3).Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(resultQuery)).To(BeEquivalentTo(4)) + Expect(resultQuery[0]).To(BeTrue()) + Expect(resultQuery[1]).To(BeTrue()) + Expect(resultQuery[2]).To(BeFalse()) + Expect(resultQuery[3]).To(BeTrue()) + + resultCount, err := client.TopKCount(ctx, "topk1", "item1", "item2", "item3").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(resultCount)).To(BeEquivalentTo(3)) + Expect(resultCount[0]).To(BeEquivalentTo(int64(2))) + Expect(resultCount[1]).To(BeEquivalentTo(int64(1))) + Expect(resultCount[2]).To(BeEquivalentTo(int64(0))) + + resultIncr, err := client.TopKIncrBy(ctx, "topk1", "item1", 5, "item2", 10).Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(resultIncr)).To(BeEquivalentTo(2)) + + resultCount, err = client.TopKCount(ctx, "topk1", "item1", "item2", "item3").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(resultCount)).To(BeEquivalentTo(3)) + Expect(resultCount[0]).To(BeEquivalentTo(int64(7))) + Expect(resultCount[1]).To(BeEquivalentTo(int64(11))) + Expect(resultCount[2]).To(BeEquivalentTo(int64(0))) + + resultList, err := client.TopKList(ctx, "topk1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(resultList)).To(BeEquivalentTo(3)) + Expect(resultList).To(ContainElements("item2", "item1", "3")) + + resultListWithCount, err := client.TopKListWithCount(ctx, "topk1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(resultListWithCount)).To(BeEquivalentTo(3)) + Expect(resultListWithCount["3"]).To(BeEquivalentTo(int64(1))) + Expect(resultListWithCount["item1"]).To(BeEquivalentTo(int64(7))) + Expect(resultListWithCount["item2"]).To(BeEquivalentTo(int64(11))) + }) + + It("should TopKReserveWithOptions", Label("topk", "topkreservewithoptions"), func() { + err := client.TopKReserveWithOptions(ctx, "topk1", 3, 1500, 8, 0.5).Err() + Expect(err).NotTo(HaveOccurred()) + + resultInfo, err := client.TopKInfo(ctx, "topk1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(resultInfo.K).To(BeEquivalentTo(int64(3))) + Expect(resultInfo.Width).To(BeEquivalentTo(int64(1500))) + Expect(resultInfo.Depth).To(BeEquivalentTo(int64(8))) + Expect(resultInfo.Decay).To(BeEquivalentTo(0.5)) + }) + + }) + + Describe("t-digest", Label("tdigest"), func() { + It("should TDigestAdd, TDigestCreate, TDigestInfo, TDigestByRank, TDigestByRevRank, TDigestCDF, TDigestMax, TDigestMin, TDigestQuantile, TDigestRank, TDigestRevRank, TDigestTrimmedMean, TDigestReset, ", Label("tdigest", "tdigestadd", "tdigestcreate", "tdigestinfo", "tdigestbyrank", "tdigestbyrevrank", "tdigestcdf", "tdigestmax", "tdigestmin", "tdigestquantile", "tdigestrank", "tdigestrevrank", "tdigesttrimmedmean", "tdigestreset"), func() { + err := client.TDigestCreate(ctx, "tdigest1").Err() + Expect(err).NotTo(HaveOccurred()) + + info, err := client.TDigestInfo(ctx, "tdigest1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(info.Observations).To(BeEquivalentTo(int64(0))) + + // Test with empty sketch + byRank, err := client.TDigestByRank(ctx, "tdigest1", 0, 1, 2, 3).Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(byRank)).To(BeEquivalentTo(4)) + + byRevRank, err := client.TDigestByRevRank(ctx, "tdigest1", 0, 1, 2).Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(byRevRank)).To(BeEquivalentTo(3)) + + cdf, err := client.TDigestCDF(ctx, "tdigest1", 15, 35, 70).Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(cdf)).To(BeEquivalentTo(3)) + + max, err := client.TDigestMax(ctx, "tdigest1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(math.IsNaN(max)).To(BeTrue()) + + min, err := client.TDigestMin(ctx, "tdigest1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(math.IsNaN(min)).To(BeTrue()) + + quantile, err := client.TDigestQuantile(ctx, "tdigest1", 0.1, 0.2).Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(quantile)).To(BeEquivalentTo(2)) + + rank, err := client.TDigestRank(ctx, "tdigest1", 10, 20).Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(rank)).To(BeEquivalentTo(2)) + + revRank, err := client.TDigestRevRank(ctx, "tdigest1", 10, 20).Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(revRank)).To(BeEquivalentTo(2)) + + trimmedMean, err := client.TDigestTrimmedMean(ctx, "tdigest1", 0.1, 0.6).Result() + Expect(err).NotTo(HaveOccurred()) + Expect(math.IsNaN(trimmedMean)).To(BeTrue()) + + // Add elements + err = client.TDigestAdd(ctx, "tdigest1", 10, 20, 30, 40, 50, 60, 70, 80, 90, 100).Err() + Expect(err).NotTo(HaveOccurred()) + + info, err = client.TDigestInfo(ctx, "tdigest1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(info.Observations).To(BeEquivalentTo(int64(10))) + + byRank, err = client.TDigestByRank(ctx, "tdigest1", 0, 1, 2).Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(byRank)).To(BeEquivalentTo(3)) + Expect(byRank[0]).To(BeEquivalentTo(float64(10))) + Expect(byRank[1]).To(BeEquivalentTo(float64(20))) + Expect(byRank[2]).To(BeEquivalentTo(float64(30))) + + byRevRank, err = client.TDigestByRevRank(ctx, "tdigest1", 0, 1, 2).Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(byRevRank)).To(BeEquivalentTo(3)) + Expect(byRevRank[0]).To(BeEquivalentTo(float64(100))) + Expect(byRevRank[1]).To(BeEquivalentTo(float64(90))) + Expect(byRevRank[2]).To(BeEquivalentTo(float64(80))) + + cdf, err = client.TDigestCDF(ctx, "tdigest1", 15, 35, 70).Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(cdf)).To(BeEquivalentTo(3)) + Expect(cdf[0]).To(BeEquivalentTo(0.1)) + Expect(cdf[1]).To(BeEquivalentTo(0.3)) + Expect(cdf[2]).To(BeEquivalentTo(0.65)) + + max, err = client.TDigestMax(ctx, "tdigest1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(max).To(BeEquivalentTo(float64(100))) + + min, err = client.TDigestMin(ctx, "tdigest1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(min).To(BeEquivalentTo(float64(10))) + + quantile, err = client.TDigestQuantile(ctx, "tdigest1", 0.1, 0.2).Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(quantile)).To(BeEquivalentTo(2)) + Expect(quantile[0]).To(BeEquivalentTo(float64(20))) + Expect(quantile[1]).To(BeEquivalentTo(float64(30))) + + rank, err = client.TDigestRank(ctx, "tdigest1", 10, 20).Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(rank)).To(BeEquivalentTo(2)) + Expect(rank[0]).To(BeEquivalentTo(int64(0))) + Expect(rank[1]).To(BeEquivalentTo(int64(1))) + + revRank, err = client.TDigestRevRank(ctx, "tdigest1", 10, 20).Result() + Expect(err).NotTo(HaveOccurred()) + Expect(len(revRank)).To(BeEquivalentTo(2)) + Expect(revRank[0]).To(BeEquivalentTo(int64(9))) + Expect(revRank[1]).To(BeEquivalentTo(int64(8))) + + trimmedMean, err = client.TDigestTrimmedMean(ctx, "tdigest1", 0.1, 0.6).Result() + Expect(err).NotTo(HaveOccurred()) + Expect(trimmedMean).To(BeEquivalentTo(float64(40))) + + reset, err := client.TDigestReset(ctx, "tdigest1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(reset).To(BeEquivalentTo("OK")) + + }) + + It("should TDigestCreateWithCompression", Label("tdigest", "tcreatewithcompression"), func() { + err := client.TDigestCreateWithCompression(ctx, "tdigest1", 2000).Err() + Expect(err).NotTo(HaveOccurred()) + + info, err := client.TDigestInfo(ctx, "tdigest1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(info.Compression).To(BeEquivalentTo(int64(2000))) + }) + + It("should TDigestMerge", Label("tdigest", "tmerge"), func() { + err := client.TDigestCreate(ctx, "tdigest1").Err() + Expect(err).NotTo(HaveOccurred()) + err = client.TDigestAdd(ctx, "tdigest1", 10, 20, 30, 40, 50, 60, 70, 80, 90, 100).Err() + Expect(err).NotTo(HaveOccurred()) + + err = client.TDigestCreate(ctx, "tdigest2").Err() + Expect(err).NotTo(HaveOccurred()) + err = client.TDigestAdd(ctx, "tdigest2", 15, 25, 35, 45, 55, 65, 75, 85, 95, 105).Err() + Expect(err).NotTo(HaveOccurred()) + + err = client.TDigestCreate(ctx, "tdigest3").Err() + Expect(err).NotTo(HaveOccurred()) + err = client.TDigestAdd(ctx, "tdigest3", 50, 60, 70, 80, 90, 100, 110, 120, 130, 140).Err() + Expect(err).NotTo(HaveOccurred()) + + options := &redis.TDigestMergeOptions{ + Compression: 1000, + Override: false, + } + err = client.TDigestMerge(ctx, "tdigest1", options, "tdigest2", "tdigest3").Err() + Expect(err).NotTo(HaveOccurred()) + + info, err := client.TDigestInfo(ctx, "tdigest1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(info.Observations).To(BeEquivalentTo(int64(30))) + Expect(info.Compression).To(BeEquivalentTo(int64(1000))) + + max, err := client.TDigestMax(ctx, "tdigest1").Result() + Expect(err).NotTo(HaveOccurred()) + Expect(max).To(BeEquivalentTo(float64(140))) + }) + }) +})