Recover from a panic during query execution

CHANGELOG.md

@@ -26,6 +26,7 @@
 - [#6379](https://github.com/influxdata/influxdb/issues/6379): Validate the first argument to percentile() is a variable.
 - [#6294](https://github.com/influxdata/influxdb/issues/6294): Fix panic running influx_inspect info.
 - [#6382](https://github.com/influxdata/influxdb/pull/6382): Removed dead code from the old query engine.
+- [#6383](https://github.com/influxdata/influxdb/pull/6383): Recover from a panic during query execution.
 
 ## v0.12.0 [2016-04-05]
 ### Release Notes

influxql/query_executor.go

@@ -156,6 +156,7 @@ func (e *QueryExecutor) ExecuteQuery(query *Query, database string, chunkSize in
 
 func (e *QueryExecutor) executeQuery(query *Query, database string, chunkSize int, closing <-chan struct{}, results chan *Result) {
 	defer close(results)
+	defer e.recover(query, results)
 
 	e.statMap.Add(statQueriesActive, 1)
 	defer func(start time.Time) {
@@ -273,6 +274,15 @@ loop:
 	}
 }
 
+func (e *QueryExecutor) recover(query *Query, results chan *Result) {
+	if err := recover(); err != nil {
+		results <- &Result{
+			StatementID: -1,
+			Err:         fmt.Errorf("%s [panic:%s]", query.String(), err),
+		}
+	}
+}
+
 func (e *QueryExecutor) executeKillQueryStatement(stmt *KillQueryStatement) error {
 	return e.killQuery(stmt.QueryID)
 }

influxql/query_executor_test.go

@@ -250,6 +250,29 @@ func TestQueryExecutor_Close(t *testing.T) {
 	}
 }
 
+func TestQueryExecutor_Panic(t *testing.T) {
+	q, err := influxql.ParseQuery(`SELECT count(value) FROM cpu`)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	e := influxql.NewQueryExecutor()
+	e.StatementExecutor = &StatementExecutor{
+		ExecuteStatementFn: func(stmt influxql.Statement, ctx *influxql.ExecutionContext) error {
+			panic("test error")
+		},
+	}
+
+	results := e.ExecuteQuery(q, "mydb", 100, nil)
+	result := <-results
+	if len(result.Series) != 0 {
+		t.Errorf("expected %d rows, got %d", 0, len(result.Series))
+	}
+	if result.Err == nil || result.Err.Error() != "SELECT count(value) FROM cpu [panic:test error]" {
+		t.Errorf("unexpected error: %s", result.Err)
+	}
+}
+
 func discardOutput(results <-chan *influxql.Result) {
 	for range results {
 		// Read all results and discard.