Merge pull request #2548 from influxdb/numeric_agg_check

Numeric aggregation check

CHANGELOG.md

@@ -20,6 +20,7 @@ This release has a breaking API change for writes -- the field previously called
 - [#2539](https://github.com/influxdb/influxdb/issues/2539): Add additional vote request logging.
 - [#2541](https://github.com/influxdb/influxdb/issues/2541): Update messaging client connection index with every message.
 - [#2542](https://github.com/influxdb/influxdb/issues/2542): Throw parser error for invalid aggregate without where time.
+- [#2548](https://github.com/influxdb/influxdb/issues/2548): Return an error when numeric aggregate applied to non-numeric data.
 
 ## v0.9.0-rc29 [2015-05-05]
 

cmd/influxd/server_integration_test.go

@@ -720,6 +720,31 @@ func runTestsData(t *testing.T, testName string, nodes Cluster, database, retent
 			queryDb:  "%DB%",
 			expected: `{"results":[{"series":[{"name":"cpu","columns":["time","sum"],"values":[["1970-01-01T00:00:00Z",50]]}]}]}`,
 		},
+		{
+			reset: true,
+			name:  "numeric aggregates on non-numeric data.",
+			write: `{"database" : "%DB%", "retentionPolicy" : "%RP%", "points": [
+				{"name": "cpu", "time": "2015-04-20T14:27:41Z", "fields": {"value": "hello"}}
+			]}`,
+			query:    `SELECT STDDEV(value) FROM cpu`,
+			queryDb:  "%DB%",
+			expected: `{"results":[{"error":"aggregate 'stddev' requires numerical field values. Field 'value' is of type string"}]}`,
+		},
+		{
+			query:    `SELECT mean(value) FROM cpu`,
+			queryDb:  "%DB%",
+			expected: `{"results":[{"error":"aggregate 'mean' requires numerical field values. Field 'value' is of type string"}]}`,
+		},
+		{
+			query:    `SELECT median(value) FROM cpu`,
+			queryDb:  "%DB%",
+			expected: `{"results":[{"error":"aggregate 'median' requires numerical field values. Field 'value' is of type string"}]}`,
+		},
+		{
+			query:    `SELECT count(value) FROM cpu`,
+			queryDb:  "%DB%",
+			expected: `{"results":[{"series":[{"name":"cpu","columns":["time","count"],"values":[["1970-01-01T00:00:00Z",1]]}]}]}`,
+		},
 
 		// Precision-specified writes
 		{

influxql/functions.go

@@ -749,6 +749,16 @@ func ReducePercentile(percentile float64) ReduceFunc {
 	}
 }
 
+// IsNumeric returns whether a given aggregate can only be run on numeric fields.
+func IsNumeric(c *Call) bool {
+	switch c.Name {
+	case "count", "first", "last":
+		return false
+	default:
+		return true
+	}
+}
+
 // MapRawQuery is for queries without aggregates
 func MapRawQuery(itr Iterator) interface{} {
 	var values []*rawQueryMapOutput

tx.go

@@ -93,6 +93,21 @@ func (tx *tx) CreateMapReduceJobs(stmt *influxql.SelectStatement, tagKeys []stri
 			}
 		}
 
+		// If a numerical aggregate is requested, ensure it is only performed on numeric data.
+		for _, a := range stmt.FunctionCalls() {
+			lit, ok := a.Args[0].(*influxql.VarRef)
+			if !ok {
+				return nil, fmt.Errorf("aggregate call didn't contain a field %s", a.String())
+			}
+			if influxql.IsNumeric(a) {
+				f := m.FieldByName(lit.Val)
+				if f.Type != influxql.Float && f.Type != influxql.Integer {
+					return nil, fmt.Errorf("aggregate '%s' requires numerical field values. Field '%s' is of type %s",
+						a.Name, f.Name, f.Type)
+				}
+			}
+		}
+
 		// Grab time range from statement.
 		tmin, tmax := influxql.TimeRange(stmt.Condition)
 		if tmax.IsZero() {
@@ -333,10 +348,7 @@ func (l *LocalMapper) Begin(c *influxql.Call, startingTime int64, chunkSize int)
 			l.limit = math.MaxUint64
 		}
 	} else {
-		lit, ok := c.Args[0].(*influxql.VarRef)
-		if !ok {
-			return fmt.Errorf("aggregate call didn't contain a field %s", c.String())
-		}
+		lit, _ := c.Args[0].(*influxql.VarRef)
 		fieldName = lit.Val
 	}