Merge pull request #2390 from neonstalwart/running-mean

add test for large mean aggregations

cmd/influxd/server_integration_test.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"log"
+	"math"
 	"net"
 	"net/http"
 	"net/url"
@@ -406,6 +407,7 @@ func runTestsData(t *testing.T, testName string, nodes Cluster, database, retent
 
 	yesterday := time.Now().Add(-1 * time.Hour * 24).UTC()
 	now := time.Now().UTC()
+	maxFloat64, _ := json.Marshal(math.MaxFloat64)
 
 	// Start by ensuring database and retention policy exist.
 	createDatabase(t, testName, nodes, database)
@@ -577,6 +579,17 @@ func runTestsData(t *testing.T, testName string, nodes Cluster, database, retent
 		},
 
 		// Aggregations
+		{
+			reset: true,
+			name:  "large mean",
+			write: `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [
+				{"name": "cpu", "timestamp": "2015-04-20T14:27:40Z", "fields": {"value": ` + string(maxFloat64) + `}},
+				{"name": "cpu", "timestamp": "2015-04-20T14:27:41Z", "fields": {"value": ` + string(maxFloat64) + `}}
+			]}`,
+			query:    `SELECT mean(value) FROM cpu`,
+			queryDb:  "%DB%",
+			expected: `{"results":[{"series":[{"name":"cpu","columns":["time","mean"],"values":[["1970-01-01T00:00:00Z",` + string(maxFloat64) + `]]}]}]}`,
+		},
 		{
 			reset: true,
 			name:  "aggregations",

influxql/functions.go

@@ -211,29 +211,31 @@ func MapMean(itr Iterator) interface{} {
 
 	for _, k, v := itr.Next(); k != 0; _, k, v = itr.Next() {
 		out.Count++
-		out.Sum += v.(float64)
+		out.Mean += (v.(float64) - out.Mean) / float64(out.Count)
 	}
 	return out
 }
 
 type meanMapOutput struct {
 	Count int
-	Sum   float64
+	Mean  float64
 }
 
 // ReduceMean computes the mean of values for each key.
 func ReduceMean(values []interface{}) interface{} {
 	out := &meanMapOutput{}
+	var countSum int
 	for _, v := range values {
 		if v == nil {
 			continue
 		}
 		val := v.(*meanMapOutput)
-		out.Count += val.Count
-		out.Sum += val.Sum
+		countSum = out.Count + val.Count
+		out.Mean = val.Mean*(float64(val.Count)/float64(countSum)) + out.Mean*(float64(out.Count)/float64(countSum))
+		out.Count = countSum
 	}
 	if out.Count > 0 {
-		return out.Sum / float64(out.Count)
+		return out.Mean
 	}
 	return nil
 }