fix: handle rate operators for table panel (#7695)

* fix: handle rate operators for table panel

pkg/query-service/app/logs/v3/query_builder.go

@@ -26,6 +26,7 @@ var AggregateOperatorToSQLFunc = map[v3.AggregateOperator]string{
 	v3.AggregateOperatorMax:     "max",
 	v3.AggregateOperatorMin:     "min",
 	v3.AggregateOperatorSum:     "sum",
+	v3.AggregateOperatorRate:    "count",
 	v3.AggregateOperatorRateSum: "sum",
 	v3.AggregateOperatorRateAvg: "avg",
 	v3.AggregateOperatorRateMax: "max",

pkg/query-service/app/logs/v4/query_builder.go

@@ -282,7 +282,7 @@ func orderByAttributeKeyTags(panelType v3.PanelType, items []v3.OrderBy, tags []
 	return str
 }
 
-func generateAggregateClause(aggOp v3.AggregateOperator,
+func generateAggregateClause(panelType v3.PanelType, start, end int64, aggOp v3.AggregateOperator,
 	aggKey string,
 	step int64,
 	timeFilter string,
@@ -296,18 +296,20 @@ func generateAggregateClause(aggOp v3.AggregateOperator,
 		"%s%s" +
 		"%s"
 	switch aggOp {
-	case v3.AggregateOperatorRate:
-		rate := float64(step)
-
-		op := fmt.Sprintf("count(%s)/%f", aggKey, rate)
-		query := fmt.Sprintf(queryTmpl, op, whereClause, groupBy, having, orderBy)
-		return query, nil
 	case
 		v3.AggregateOperatorRateSum,
 		v3.AggregateOperatorRateMax,
 		v3.AggregateOperatorRateAvg,
-		v3.AggregateOperatorRateMin:
+		v3.AggregateOperatorRateMin,
+		v3.AggregateOperatorRate:
 		rate := float64(step)
+		if panelType == v3.PanelTypeTable {
+			// if the panel type is table the denominator will be the total time range
+			duration := end - start
+			if duration >= 0 {
+				rate = float64(duration) / NANOSECOND
+			}
+		}
 
 		op := fmt.Sprintf("%s(%s)/%f", logsV3.AggregateOperatorToSQLFunc[aggOp], aggKey, rate)
 		query := fmt.Sprintf(queryTmpl, op, whereClause, groupBy, having, orderBy)
@@ -418,7 +420,7 @@ func buildLogsQuery(panelType v3.PanelType, start, end, step int64, mq *v3.Build
 		filterSubQuery = filterSubQuery + " AND " + fmt.Sprintf("(%s) GLOBAL IN (", logsV3.GetSelectKeys(mq.AggregateOperator, mq.GroupBy)) + "#LIMIT_PLACEHOLDER)"
 	}
 
-	aggClause, err := generateAggregateClause(mq.AggregateOperator, aggregationKey, step, timeFilter, filterSubQuery, groupBy, having, orderBy)
+	aggClause, err := generateAggregateClause(panelType, logsStart, logsEnd, mq.AggregateOperator, aggregationKey, step, timeFilter, filterSubQuery, groupBy, having, orderBy)
 	if err != nil {
 		return "", err
 	}

pkg/query-service/app/logs/v4/query_builder_test.go

@@ -571,6 +571,9 @@ func Test_orderByAttributeKeyTags(t *testing.T) {
 
 func Test_generateAggregateClause(t *testing.T) {
 	type args struct {
+		panelType   v3.PanelType
+		start       int64
+		end         int64
 		op          v3.AggregateOperator
 		aggKey      string
 		step        int64
@@ -590,6 +593,7 @@ func Test_generateAggregateClause(t *testing.T) {
 			name: "test rate",
 			args: args{
 				op:          v3.AggregateOperatorRate,
+				panelType:   v3.PanelTypeGraph,
 				aggKey:      "test",
 				step:        60,
 				timeFilter:  "(timestamp >= 1680066360726210000 AND timestamp <= 1680066458000000000) AND (ts_bucket_start >= 1680064560 AND ts_bucket_start <= 1680066458)",
@@ -606,6 +610,7 @@ func Test_generateAggregateClause(t *testing.T) {
 			name: "test P10 with all args",
 			args: args{
 				op:          v3.AggregateOperatorRate,
+				panelType:   v3.PanelTypeGraph,
 				aggKey:      "test",
 				step:        60,
 				timeFilter:  "(timestamp >= 1680066360726210000 AND timestamp <= 1680066458000000000) AND (ts_bucket_start >= 1680064560 AND ts_bucket_start <= 1680066458)",
@@ -618,10 +623,29 @@ func Test_generateAggregateClause(t *testing.T) {
 				"(ts_bucket_start >= 1680064560 AND ts_bucket_start <= 1680066458) AND attributes_string['service.name'] = 'test' group by `user_name` having value > 10 order by " +
 				"`user_name` desc",
 		},
+		{
+			name: "test rate for table panel",
+			args: args{
+				op:          v3.AggregateOperatorRate,
+				panelType:   v3.PanelTypeTable,
+				start:       1745315470000000000,
+				end:         1745319070000000000,
+				aggKey:      "test",
+				step:        60,
+				timeFilter:  "(timestamp >= 1680066360726210000 AND timestamp <= 1680066458000000000) AND (ts_bucket_start >= 1680064560 AND ts_bucket_start <= 1680066458)",
+				whereClause: " AND attributes_string['service.name'] = 'test'",
+				groupBy:     " group by `user_name`",
+				having:      "",
+				orderBy:     " order by `user_name` desc",
+			},
+			want: " count(test)/3600.000000 as value from signoz_logs.distributed_logs_v2 where (timestamp >= 1680066360726210000 AND timestamp <= 1680066458000000000) AND " +
+				"(ts_bucket_start >= 1680064560 AND ts_bucket_start <= 1680066458) AND attributes_string['service.name'] = 'test' " +
+				"group by `user_name` order by `user_name` desc",
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := generateAggregateClause(tt.args.op, tt.args.aggKey, tt.args.step, tt.args.timeFilter, tt.args.whereClause, tt.args.groupBy, tt.args.having, tt.args.orderBy)
+			got, err := generateAggregateClause(tt.args.panelType, tt.args.start, tt.args.end, tt.args.op, tt.args.aggKey, tt.args.step, tt.args.timeFilter, tt.args.whereClause, tt.args.groupBy, tt.args.having, tt.args.orderBy)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("generateAggreagteClause() error = %v, wantErr %v", err, tt.wantErr)
 				return

pkg/query-service/app/traces/v4/query_builder.go

@@ -367,8 +367,15 @@ func buildTracesQuery(start, end, step int64, mq *v3.BuilderQuery, panelType v3.
 		v3.AggregateOperatorRateAvg,
 		v3.AggregateOperatorRateMin,
 		v3.AggregateOperatorRate:
-
 		rate := float64(step)
+		if panelType == v3.PanelTypeTable {
+			// if the panel type is table the denominator will be the total time range
+			duration := tracesEnd - tracesStart
+			if duration >= 0 {
+				rate = float64(duration) / NANOSECOND
+			}
+		}
+
 		op := fmt.Sprintf("%s(%s)/%f", tracesV3.AggregateOperatorToSQLFunc[mq.AggregateOperator], aggregationKey, rate)
 		query := fmt.Sprintf(queryTmpl, op, filterSubQuery, groupBy, having, orderBy)
 		return query, nil

pkg/query-service/app/traces/v4/query_builder_test.go

@@ -693,6 +693,38 @@ func Test_buildTracesQuery(t *testing.T) {
 			want: "SELECT  max(toUnixTimestamp64Nano(timestamp)) as value from signoz_traces.distributed_signoz_index_v3 where (timestamp >= '1680066360726210000' AND timestamp <= '1680066458000000000') AND " +
 				"(ts_bucket_start >= 1680064560 AND ts_bucket_start <= 1680066458) order by value DESC",
 		},
+		{
+			name: "test rate for graph panel",
+			args: args{
+				panelType: v3.PanelTypeGraph,
+				start:     1745315470000000000,
+				end:       1745319070000000000,
+				step:      60,
+				mq: &v3.BuilderQuery{
+					AggregateOperator: v3.AggregateOperatorRate,
+					StepInterval:      60,
+					Filters:           &v3.FilterSet{},
+				},
+			},
+			want: "SELECT toStartOfInterval(timestamp, INTERVAL 60 SECOND) AS ts, count()/60.000000 as value from signoz_traces.distributed_signoz_index_v3 where (timestamp >= '1745315470000000000' AND " +
+				"timestamp <= '1745319070000000000') AND (ts_bucket_start >= 1745313670 AND ts_bucket_start <= 1745319070) group by ts order by value DESC",
+		},
+		{
+			name: "test rate for table panel",
+			args: args{
+				panelType: v3.PanelTypeTable,
+				start:     1745315470000000000,
+				end:       1745319070000000000,
+				step:      60,
+				mq: &v3.BuilderQuery{
+					AggregateOperator: v3.AggregateOperatorRate,
+					StepInterval:      60,
+					Filters:           &v3.FilterSet{},
+				},
+			},
+			want: "SELECT  count()/3600.000000 as value from signoz_traces.distributed_signoz_index_v3 where (timestamp >= '1745315470000000000' AND " +
+				"timestamp <= '1745319070000000000') AND (ts_bucket_start >= 1745313670 AND ts_bucket_start <= 1745319070) order by value DESC",
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {