feat(command): Adds support for multiple aggregators in TS range commands (#3791)

* Adds support for multiple aggregators in TS range commands

* Validates against invalid aggregators

---------

Co-authored-by: Nedyalko Dyakov <[email protected]>
This commit is contained in:
Elena Kolevska
2026-05-21 14:10:55 +03:00
committed by GitHub
co-authored by Nedyalko Dyakov
parent 4653645e9d
commit 540781831b
3 changed files with 392 additions and 41 deletions
+20
View File
@@ -669,8 +669,28 @@ func ExampleClient_timeseries_aggregation() {
// >>> [{0 1.9500000000000002} {2 2.0999999999999996} {4 1.78}]
// STEP_END
// STEP_START agg_multi
res33, err := rdb.TSRangeWithArgs(
ctx,
"rg:2",
0,
math.MaxInt64,
&redis.TSRangeOptions{
Aggregators: []redis.Aggregator{redis.Min, redis.Max},
BucketDuration: 2,
},
).Result()
if err != nil {
panic(err)
}
fmt.Println(res33)
// >>> [{0 [1.8 2.1]} {2 [1.9 2.3]} {4 [1.78 1.78]}]
// STEP_END
// Output:
// [{0 1.9500000000000002} {2 2.0999999999999996} {4 1.78}]
// [{0 [1.8 2.1]} {2 [1.9 2.3]} {4 [1.78 1.78]}]
}
func ExampleClient_timeseries_agg_bucket() {
ctx := context.Background()
+172 -41
View File
@@ -2,6 +2,9 @@ package redis
import (
"context"
"errors"
"fmt"
"strings"
"github.com/redis/go-redis/v9/internal/proto"
"github.com/redis/go-redis/v9/internal/util"
@@ -139,39 +142,88 @@ func (a Aggregator) String() string {
}
}
var (
errTSMultiAggregationGroupBy = errors.New("redis: GROUPBY is not allowed when multiple aggregators are specified")
errTSAggregationConflict = errors.New("redis: setting both Aggregator and Aggregators is not allowed; use Aggregators instead because Aggregator is deprecated")
)
func formatAggregationArgs(aggregator Aggregator, aggregators []Aggregator) (string, int, error) {
if aggregator != Invalid && len(aggregators) > 0 {
return "", 0, errTSAggregationConflict
}
if len(aggregators) == 0 {
if aggregator == Invalid {
return "", 0, nil
}
aggregationArg, err := formatAggregatorArg(aggregator)
if err != nil {
return "", 0, err
}
return aggregationArg, 1, nil
}
parts := make([]string, len(aggregators))
for i, agg := range aggregators {
if agg == Invalid {
return "", 0, fmt.Errorf("redis: invalid timeseries aggregator at index %d: Invalid (%d)", i, agg)
}
aggregationArg, err := formatAggregatorArg(agg)
if err != nil {
return "", 0, fmt.Errorf("redis: invalid timeseries aggregator at index %d: %d", i, agg)
}
parts[i] = aggregationArg
}
return strings.Join(parts, ","), len(parts), nil
}
func formatAggregatorArg(aggregator Aggregator) (string, error) {
aggregationArg := aggregator.String()
if aggregationArg == "" {
return "", fmt.Errorf("redis: invalid timeseries aggregator: %d", aggregator)
}
return aggregationArg, nil
}
type TSRangeOptions struct {
Latest bool
FilterByTS []int
FilterByValue []int
Count int
Align interface{}
Latest bool
FilterByTS []int
FilterByValue []int
Count int
Align interface{}
// Deprecated: use Aggregators instead.
Aggregator Aggregator
Aggregators []Aggregator
BucketDuration int
BucketTimestamp interface{}
Empty bool
}
type TSRevRangeOptions struct {
Latest bool
FilterByTS []int
FilterByValue []int
Count int
Align interface{}
Latest bool
FilterByTS []int
FilterByValue []int
Count int
Align interface{}
// Deprecated: use Aggregators instead.
Aggregator Aggregator
Aggregators []Aggregator
BucketDuration int
BucketTimestamp interface{}
Empty bool
}
type TSMRangeOptions struct {
Latest bool
FilterByTS []int
FilterByValue []int
WithLabels bool
SelectedLabels []interface{}
Count int
Align interface{}
Latest bool
FilterByTS []int
FilterByValue []int
WithLabels bool
SelectedLabels []interface{}
Count int
Align interface{}
// Deprecated: use Aggregators instead.
Aggregator Aggregator
Aggregators []Aggregator
BucketDuration int
BucketTimestamp interface{}
Empty bool
@@ -180,14 +232,16 @@ type TSMRangeOptions struct {
}
type TSMRevRangeOptions struct {
Latest bool
FilterByTS []int
FilterByValue []int
WithLabels bool
SelectedLabels []interface{}
Count int
Align interface{}
Latest bool
FilterByTS []int
FilterByValue []int
WithLabels bool
SelectedLabels []interface{}
Count int
Align interface{}
// Deprecated: use Aggregators instead.
Aggregator Aggregator
Aggregators []Aggregator
BucketDuration int
BucketTimestamp interface{}
Empty bool
@@ -483,7 +537,16 @@ func (c cmdable) TSGet(ctx context.Context, key string) *TSTimestampValueCmd {
type TSTimestampValue struct {
Timestamp int64
Value float64
Values []float64
}
func (tv TSTimestampValue) String() string {
if len(tv.Values) > 0 {
return fmt.Sprintf("{%d %v}", tv.Timestamp, tv.Values)
}
return fmt.Sprintf("{%d %v}", tv.Timestamp, tv.Value)
}
type TSTimestampValueCmd struct {
baseCmd
val TSTimestampValue
@@ -541,9 +604,14 @@ func (cmd *TSTimestampValueCmd) readReply(rd *proto.Reader) (err error) {
}
func (cmd *TSTimestampValueCmd) Clone() Cmder {
val := cmd.val
if cmd.val.Values != nil {
val.Values = make([]float64, len(cmd.val.Values))
copy(val.Values, cmd.val.Values)
}
return &TSTimestampValueCmd{
baseCmd: cmd.cloneBaseCmd(),
val: cmd.val, // TSTimestampValue is a simple struct, can be copied directly
val: val,
}
}
@@ -636,8 +704,14 @@ func (c cmdable) TSRevRangeWithArgs(ctx context.Context, key string, fromTimesta
if options.Align != nil {
args = append(args, "ALIGN", options.Align)
}
if options.Aggregator != 0 {
args = append(args, "AGGREGATION", options.Aggregator.String())
aggregationArg, _, err := formatAggregationArgs(options.Aggregator, options.Aggregators)
if err != nil {
cmd := newTSTimestampValueSliceCmd(ctx, args...)
cmd.SetErr(err)
return cmd
}
if aggregationArg != "" {
args = append(args, "AGGREGATION", aggregationArg)
}
if options.BucketDuration != 0 {
args = append(args, options.BucketDuration)
@@ -692,8 +766,14 @@ func (c cmdable) TSRangeWithArgs(ctx context.Context, key string, fromTimestamp
if options.Align != nil {
args = append(args, "ALIGN", options.Align)
}
if options.Aggregator != 0 {
args = append(args, "AGGREGATION", options.Aggregator.String())
aggregationArg, _, err := formatAggregationArgs(options.Aggregator, options.Aggregators)
if err != nil {
cmd := newTSTimestampValueSliceCmd(ctx, args...)
cmd.SetErr(err)
return cmd
}
if aggregationArg != "" {
args = append(args, "AGGREGATION", aggregationArg)
}
if options.BucketDuration != 0 {
args = append(args, options.BucketDuration)
@@ -748,19 +828,38 @@ func (cmd *TSTimestampValueSliceCmd) readReply(rd *proto.Reader) (err error) {
}
cmd.val = make([]TSTimestampValue, n)
for i := 0; i < n; i++ {
_, _ = rd.ReadArrayLen()
itemLen, err := rd.ReadArrayLen()
if err != nil {
return err
}
timestamp, err := rd.ReadInt()
if err != nil {
return err
}
value, err := rd.ReadString()
if err != nil {
return err
}
cmd.val[i].Timestamp = timestamp
cmd.val[i].Value, err = util.ParseStringToFloat(value)
if err != nil {
return err
if itemLen == 2 {
value, err := rd.ReadString()
if err != nil {
return err
}
cmd.val[i].Value, err = util.ParseStringToFloat(value)
if err != nil {
return err
}
continue
}
cmd.val[i].Values = make([]float64, itemLen-1)
for j := 0; j < itemLen-1; j++ {
value, err := rd.ReadString()
if err != nil {
return err
}
cmd.val[i].Values[j], err = util.ParseStringToFloat(value)
if err != nil {
return err
}
}
}
@@ -772,6 +871,12 @@ func (cmd *TSTimestampValueSliceCmd) Clone() Cmder {
if cmd.val != nil {
val = make([]TSTimestampValue, len(cmd.val))
copy(val, cmd.val)
for i := range cmd.val {
if cmd.val[i].Values != nil {
val[i].Values = make([]float64, len(cmd.val[i].Values))
copy(val[i].Values, cmd.val[i].Values)
}
}
}
return &TSTimestampValueSliceCmd{
baseCmd: cmd.cloneBaseCmd(),
@@ -799,6 +904,7 @@ func (c cmdable) TSMRange(ctx context.Context, fromTimestamp int, toTimestamp in
// For more information - https://redis.io/commands/ts.mrange/
func (c cmdable) TSMRangeWithArgs(ctx context.Context, fromTimestamp int, toTimestamp int, filterExpr []string, options *TSMRangeOptions) *MapStringSliceInterfaceCmd {
args := []interface{}{"TS.MRANGE", fromTimestamp, toTimestamp}
multiAggregationCount := 0
if options != nil {
if options.Latest {
args = append(args, "LATEST")
@@ -828,8 +934,15 @@ func (c cmdable) TSMRangeWithArgs(ctx context.Context, fromTimestamp int, toTime
if options.Align != nil {
args = append(args, "ALIGN", options.Align)
}
if options.Aggregator != 0 {
args = append(args, "AGGREGATION", options.Aggregator.String())
aggregationArg, count, err := formatAggregationArgs(options.Aggregator, options.Aggregators)
if err != nil {
cmd := NewMapStringSliceInterfaceCmd(ctx, args...)
cmd.SetErr(err)
return cmd
}
multiAggregationCount = count
if aggregationArg != "" {
args = append(args, "AGGREGATION", aggregationArg)
}
if options.BucketDuration != 0 {
args = append(args, options.BucketDuration)
@@ -846,6 +959,11 @@ func (c cmdable) TSMRangeWithArgs(ctx context.Context, fromTimestamp int, toTime
args = append(args, f)
}
if options != nil {
if multiAggregationCount > 1 && (options.GroupByLabel != nil || options.Reducer != nil) {
cmd := NewMapStringSliceInterfaceCmd(ctx, args...)
cmd.SetErr(errTSMultiAggregationGroupBy)
return cmd
}
if options.GroupByLabel != nil {
args = append(args, "GROUPBY", options.GroupByLabel)
}
@@ -878,6 +996,7 @@ func (c cmdable) TSMRevRange(ctx context.Context, fromTimestamp int, toTimestamp
// For more information - https://redis.io/commands/ts.mrevrange/
func (c cmdable) TSMRevRangeWithArgs(ctx context.Context, fromTimestamp int, toTimestamp int, filterExpr []string, options *TSMRevRangeOptions) *MapStringSliceInterfaceCmd {
args := []interface{}{"TS.MREVRANGE", fromTimestamp, toTimestamp}
multiAggregationCount := 0
if options != nil {
if options.Latest {
args = append(args, "LATEST")
@@ -907,8 +1026,15 @@ func (c cmdable) TSMRevRangeWithArgs(ctx context.Context, fromTimestamp int, toT
if options.Align != nil {
args = append(args, "ALIGN", options.Align)
}
if options.Aggregator != 0 {
args = append(args, "AGGREGATION", options.Aggregator.String())
aggregationArg, count, err := formatAggregationArgs(options.Aggregator, options.Aggregators)
if err != nil {
cmd := NewMapStringSliceInterfaceCmd(ctx, args...)
cmd.SetErr(err)
return cmd
}
multiAggregationCount = count
if aggregationArg != "" {
args = append(args, "AGGREGATION", aggregationArg)
}
if options.BucketDuration != 0 {
args = append(args, options.BucketDuration)
@@ -925,6 +1051,11 @@ func (c cmdable) TSMRevRangeWithArgs(ctx context.Context, fromTimestamp int, toT
args = append(args, f)
}
if options != nil {
if multiAggregationCount > 1 && (options.GroupByLabel != nil || options.Reducer != nil) {
cmd := NewMapStringSliceInterfaceCmd(ctx, args...)
cmd.SetErr(errTSMultiAggregationGroupBy)
return cmd
}
if options.GroupByLabel != nil {
args = append(args, "GROUPBY", options.GroupByLabel)
}
+200
View File
@@ -756,6 +756,59 @@ var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
Expect(len(resultRange)).To(BeEquivalentTo(7))
})
It("should TSRangeWithArgs support multiple aggregators", Label("timeseries", "tsrange", "tsrangeWithArgs", "aggregators", "NonRedisEnterprise"), func() {
SkipBeforeRedisVersion(8.8, "multiple aggregators require Redis 8.8+")
_, err := client.TSCreate(ctx, "multi-range").Result()
Expect(err).NotTo(HaveOccurred())
_, err = client.TSMAdd(ctx, [][]interface{}{
{"multi-range", 1000, 100},
{"multi-range", 1010, 110},
{"multi-range", 1020, 120},
{"multi-range", 1030, 130},
{"multi-range", 1040, 140},
{"multi-range", 1050, 150},
}).Result()
Expect(err).NotTo(HaveOccurred())
result, err := client.TSRangeWithArgs(ctx, "multi-range", 0, 2000, &redis.TSRangeOptions{
Aggregators: []redis.Aggregator{redis.Min, redis.Max},
BucketDuration: 20,
}).Result()
Expect(err).NotTo(HaveOccurred())
Expect(result).To(Equal([]redis.TSTimestampValue{
{Timestamp: 1000, Values: []float64{100, 110}},
{Timestamp: 1020, Values: []float64{120, 130}},
{Timestamp: 1040, Values: []float64{140, 150}},
}))
cmd := client.TSRangeWithArgs(ctx, "multi-range", 0, 2000, &redis.TSRangeOptions{
Aggregator: redis.Avg,
Aggregators: []redis.Aggregator{redis.Min, redis.Max},
BucketDuration: 20,
})
Expect(cmd.Err()).To(MatchError("redis: setting both Aggregator and Aggregators is not allowed; use Aggregators instead because Aggregator is deprecated"))
cmd = client.TSRangeWithArgs(ctx, "multi-range", 0, 2000, &redis.TSRangeOptions{
Aggregators: []redis.Aggregator{redis.Min, redis.Invalid, redis.Max},
BucketDuration: 20,
})
Expect(cmd.Err()).To(MatchError("redis: invalid timeseries aggregator at index 1: Invalid (0)"))
cmd = client.TSRangeWithArgs(ctx, "multi-range", 0, 2000, &redis.TSRangeOptions{
Aggregator: redis.Aggregator(999),
BucketDuration: 20,
})
Expect(cmd.Err()).To(MatchError("redis: invalid timeseries aggregator: 999"))
cmd = client.TSRangeWithArgs(ctx, "multi-range", 0, 2000, &redis.TSRangeOptions{
Aggregators: []redis.Aggregator{redis.Min, redis.Aggregator(999), redis.Max},
BucketDuration: 20,
})
Expect(cmd.Err()).To(MatchError("redis: invalid timeseries aggregator at index 1: 999"))
})
It("should TSRevRange, TSRevRangeWithArgs", Label("timeseries", "tsrevrange", "tsrevrangeWithArgs", "NonRedisEnterprise"), func() {
for i := 0; i < 100; i++ {
_, err := client.TSAdd(ctx, "a", i, float64(i%7)).Result()
@@ -884,6 +937,34 @@ var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
Expect(len(resultRange)).To(BeEquivalentTo(7))
})
It("should TSRevRangeWithArgs support multiple aggregators", Label("timeseries", "tsrevrange", "tsrevrangeWithArgs", "aggregators", "NonRedisEnterprise"), func() {
SkipBeforeRedisVersion(8.8, "multiple aggregators require Redis 8.8+")
_, err := client.TSCreate(ctx, "multi-revrange").Result()
Expect(err).NotTo(HaveOccurred())
_, err = client.TSMAdd(ctx, [][]interface{}{
{"multi-revrange", 1000, 100},
{"multi-revrange", 1010, 110},
{"multi-revrange", 1020, 120},
{"multi-revrange", 1030, 130},
{"multi-revrange", 1040, 140},
{"multi-revrange", 1050, 150},
}).Result()
Expect(err).NotTo(HaveOccurred())
result, err := client.TSRevRangeWithArgs(ctx, "multi-revrange", 0, 2000, &redis.TSRevRangeOptions{
Aggregators: []redis.Aggregator{redis.Min, redis.Max},
BucketDuration: 20,
}).Result()
Expect(err).NotTo(HaveOccurred())
Expect(result).To(Equal([]redis.TSTimestampValue{
{Timestamp: 1040, Values: []float64{140, 150}},
{Timestamp: 1020, Values: []float64{120, 130}},
{Timestamp: 1000, Values: []float64{100, 110}},
}))
})
It("should TSMRange and TSMRangeWithArgs", Label("timeseries", "tsmrange", "tsmrangeWithArgs"), func() {
createOpt := &redis.TSOptions{Labels: map[string]string{"Test": "This", "team": "ny"}}
resultCreate, err := client.TSCreateWithArgs(ctx, "a", createOpt).Result()
@@ -1071,6 +1152,65 @@ var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
Expect(result["d"][2]).To(BeEquivalentTo([]interface{}{[]interface{}{int64(0), 4.0}, []interface{}{int64(10), 8.0}}))
}
})
It("should TSMRangeWithArgs support multiple aggregators", Label("timeseries", "tsmrange", "tsmrangeWithArgs", "aggregators", "NonRedisEnterprise"), func() {
SkipBeforeRedisVersion(8.8, "multiple aggregators require Redis 8.8+")
_, err := client.TSCreateWithArgs(ctx, "multi-mrange-a", &redis.TSOptions{
Labels: map[string]string{"type": "sensor", "name": "a"},
}).Result()
Expect(err).NotTo(HaveOccurred())
_, err = client.TSCreateWithArgs(ctx, "multi-mrange-b", &redis.TSOptions{
Labels: map[string]string{"type": "sensor", "name": "b"},
}).Result()
Expect(err).NotTo(HaveOccurred())
_, err = client.TSMAdd(ctx, [][]interface{}{
{"multi-mrange-a", 1000, 10},
{"multi-mrange-a", 1010, 20},
{"multi-mrange-a", 1020, 30},
{"multi-mrange-a", 1030, 40},
{"multi-mrange-b", 1000, 15},
{"multi-mrange-b", 1010, 25},
{"multi-mrange-b", 1020, 35},
{"multi-mrange-b", 1030, 45},
}).Result()
Expect(err).NotTo(HaveOccurred())
result, err := client.TSMRangeWithArgs(ctx, 0, 2000, []string{"type=sensor"}, &redis.TSMRangeOptions{
WithLabels: true,
Aggregators: []redis.Aggregator{redis.Min, redis.Max},
BucketDuration: 20,
}).Result()
Expect(err).NotTo(HaveOccurred())
Expect(len(result)).To(Equal(2))
if client.Options().Protocol == 2 {
Expect(result["multi-mrange-a"][1]).To(Equal([]interface{}{
[]interface{}{int64(1000), "10", "20"},
[]interface{}{int64(1020), "30", "40"},
}))
Expect(result["multi-mrange-b"][1]).To(Equal([]interface{}{
[]interface{}{int64(1000), "15", "25"},
[]interface{}{int64(1020), "35", "45"},
}))
} else {
Expect(result["multi-mrange-a"][2]).To(Equal([]interface{}{
[]interface{}{int64(1000), 10.0, 20.0},
[]interface{}{int64(1020), 30.0, 40.0},
}))
Expect(result["multi-mrange-b"][2]).To(Equal([]interface{}{
[]interface{}{int64(1000), 15.0, 25.0},
[]interface{}{int64(1020), 35.0, 45.0},
}))
}
cmd := client.TSMRangeWithArgs(ctx, 0, 2000, []string{"type=sensor"}, &redis.TSMRangeOptions{
Aggregators: []redis.Aggregator{redis.Avg, redis.Count},
BucketDuration: 20,
GroupByLabel: "type",
Reducer: "max",
})
Expect(cmd.Err()).To(MatchError("redis: GROUPBY is not allowed when multiple aggregators are specified"))
})
It("should TSMRevRange and TSMRevRangeWithArgs", Label("timeseries", "tsmrevrange", "tsmrevrangeWithArgs"), func() {
createOpt := &redis.TSOptions{Labels: map[string]string{"Test": "This", "team": "ny"}}
resultCreate, err := client.TSCreateWithArgs(ctx, "a", createOpt).Result()
@@ -1199,6 +1339,66 @@ var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
}
})
It("should TSMRevRangeWithArgs support multiple aggregators", Label("timeseries", "tsmrevrange", "tsmrevrangeWithArgs", "aggregators", "NonRedisEnterprise"), func() {
SkipBeforeRedisVersion(8.8, "multiple aggregators require Redis 8.8+")
_, err := client.TSCreateWithArgs(ctx, "multi-mrevrange-a", &redis.TSOptions{
Labels: map[string]string{"type": "sensor", "name": "a"},
}).Result()
Expect(err).NotTo(HaveOccurred())
_, err = client.TSCreateWithArgs(ctx, "multi-mrevrange-b", &redis.TSOptions{
Labels: map[string]string{"type": "sensor", "name": "b"},
}).Result()
Expect(err).NotTo(HaveOccurred())
_, err = client.TSMAdd(ctx, [][]interface{}{
{"multi-mrevrange-a", 1000, 10},
{"multi-mrevrange-a", 1010, 20},
{"multi-mrevrange-a", 1020, 30},
{"multi-mrevrange-a", 1030, 40},
{"multi-mrevrange-b", 1000, 15},
{"multi-mrevrange-b", 1010, 25},
{"multi-mrevrange-b", 1020, 35},
{"multi-mrevrange-b", 1030, 45},
}).Result()
Expect(err).NotTo(HaveOccurred())
result, err := client.TSMRevRangeWithArgs(ctx, 0, 2000, []string{"type=sensor"}, &redis.TSMRevRangeOptions{
WithLabels: true,
Aggregators: []redis.Aggregator{redis.Min, redis.Max},
BucketDuration: 20,
}).Result()
Expect(err).NotTo(HaveOccurred())
Expect(len(result)).To(Equal(2))
if client.Options().Protocol == 2 {
Expect(result["multi-mrevrange-a"][1]).To(Equal([]interface{}{
[]interface{}{int64(1020), "30", "40"},
[]interface{}{int64(1000), "10", "20"},
}))
Expect(result["multi-mrevrange-b"][1]).To(Equal([]interface{}{
[]interface{}{int64(1020), "35", "45"},
[]interface{}{int64(1000), "15", "25"},
}))
} else {
Expect(result["multi-mrevrange-a"][2]).To(Equal([]interface{}{
[]interface{}{int64(1020), 30.0, 40.0},
[]interface{}{int64(1000), 10.0, 20.0},
}))
Expect(result["multi-mrevrange-b"][2]).To(Equal([]interface{}{
[]interface{}{int64(1020), 35.0, 45.0},
[]interface{}{int64(1000), 15.0, 25.0},
}))
}
cmd := client.TSMRevRangeWithArgs(ctx, 0, 2000, []string{"type=sensor"}, &redis.TSMRevRangeOptions{
Aggregators: []redis.Aggregator{redis.Avg, redis.Count},
BucketDuration: 20,
GroupByLabel: "type",
Reducer: "max",
})
Expect(cmd.Err()).To(MatchError("redis: GROUPBY is not allowed when multiple aggregators are specified"))
})
It("should TSMRevRangeWithArgs Latest", Label("timeseries", "tsmrevrangeWithArgs", "tsmrevrangelatest", "NonRedisEnterprise"), func() {
resultCreate, err := client.TSCreate(ctx, "a").Result()
Expect(err).NotTo(HaveOccurred())