From 8d35c6b2972ba6fcf65b5a3984147dd9d905c5b8 Mon Sep 17 00:00:00 2001
From: Alexander Zobnin <alexanderzobnin@gmail.com>
Date: Wed, 27 Apr 2022 15:26:48 +0300
Subject: [PATCH] Fix percentile aggregation with $__range_series interval

---
 pkg/datasource/functions.go  |  8 +++++++-
 pkg/timeseries/timeseries.go | 18 ++++++++++++++++++
 2 files changed, 25 insertions(+), 1 deletion(-)

diff --git a/pkg/datasource/functions.go b/pkg/datasource/functions.go
index 7c9c258..17012af 100644
--- a/pkg/datasource/functions.go
+++ b/pkg/datasource/functions.go
@@ -307,6 +307,13 @@ func applyPercentileAgg(series []*timeseries.TimeSeriesData, params ...interface
 	if err != nil {
 		return nil, errParsingFunctionParam(err)
 	}
+	aggFunc := timeseries.AggPercentile(percentile)
+
+	if pInterval == RANGE_VARIABLE_VALUE {
+		aggregatedSeries := timeseries.AggregateByRange(series, aggFunc)
+		aggregatedSeries.Meta.Name = fmt.Sprintf("percentileAgg(%s, %v)", pInterval, percentile)
+		return []*timeseries.TimeSeriesData{aggregatedSeries}, nil
+	}
 
 	interval, err := gtime.ParseInterval(pInterval)
 	if err != nil {
@@ -316,7 +323,6 @@ func applyPercentileAgg(series []*timeseries.TimeSeriesData, params ...interface
 		return series, nil
 	}
 
-	aggFunc := timeseries.AggPercentile(percentile)
 	aggregatedSeries := timeseries.AggregateBy(series, interval, aggFunc)
 	aggregatedSeries.Meta.Name = fmt.Sprintf("percentileAgg(%s, %v)", pInterval, percentile)
 
diff --git a/pkg/timeseries/timeseries.go b/pkg/timeseries/timeseries.go
index f265bdc..bd46ce0 100644
--- a/pkg/timeseries/timeseries.go
+++ b/pkg/timeseries/timeseries.go
@@ -166,6 +166,24 @@ func AggregateBy(series []*TimeSeriesData, interval time.Duration, aggFunc AggFu
 	return aggregatedSeriesData
 }
 
+func AggregateByRange(series []*TimeSeriesData, aggFunc AggFunc) *TimeSeriesData {
+	aggregatedSeries := NewTimeSeries()
+
+	// Combine all points into one time series
+	for _, s := range series {
+		aggregatedSeries = append(aggregatedSeries, s.TS...)
+	}
+
+	value := aggFunc(aggregatedSeries)
+	aggregatedSeriesData := NewTimeSeriesData()
+	aggregatedSeriesData.TS = []TimePoint{
+		{Time: aggregatedSeries[0].Time, Value: value},
+		{Time: aggregatedSeries[aggregatedSeries.Len()-1].Time, Value: value},
+	}
+
+	return aggregatedSeriesData
+}
+
 func (ts TimeSeries) Sort() {
 	sorted := sort.SliceIsSorted(ts, ts.less())
 	if !sorted {