Add some error checking to the engine.
parent
c962a81a5f
commit
bdc1976c2a
|
@ -0,0 +1,19 @@
|
||||||
|
package common
|
||||||
|
|
||||||
|
const (
|
||||||
|
WrongNumberOfArguments = iota
|
||||||
|
InvalidArgument
|
||||||
|
)
|
||||||
|
|
||||||
|
type QueryError struct {
|
||||||
|
ErrorCode int
|
||||||
|
ErrorMsg string
|
||||||
|
}
|
||||||
|
|
||||||
|
func (self *QueryError) Error() string {
|
||||||
|
return self.ErrorMsg
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewQueryError(code int, msg string) *QueryError {
|
||||||
|
return &QueryError{code, msg}
|
||||||
|
}
|
|
@ -105,7 +105,8 @@ type Mapper func(*protocol.Point) interface{}
|
||||||
type InverseMapper func(interface{}, int) interface{}
|
type InverseMapper func(interface{}, int) interface{}
|
||||||
|
|
||||||
func createValuesToInterface(groupBy parser.GroupByClause, definitions []*protocol.FieldDefinition) (Mapper, InverseMapper) {
|
func createValuesToInterface(groupBy parser.GroupByClause, definitions []*protocol.FieldDefinition) (Mapper, InverseMapper) {
|
||||||
window, ok := groupBy.GetGroupByTime()
|
// we shouldn't get an error, this is checked earlier in the executeCountQueryWithGroupBy
|
||||||
|
window, _ := groupBy.GetGroupByTime()
|
||||||
names := []string{}
|
names := []string{}
|
||||||
for _, value := range groupBy {
|
for _, value := range groupBy {
|
||||||
names = append(names, value.Name)
|
names = append(names, value.Name)
|
||||||
|
@ -125,8 +126,8 @@ func createValuesToInterface(groupBy parser.GroupByClause, definitions []*protoc
|
||||||
}
|
}
|
||||||
|
|
||||||
return func(p *protocol.Point) interface{} {
|
return func(p *protocol.Point) interface{} {
|
||||||
if ok {
|
if window != nil {
|
||||||
return getTimestampFromPoint(window, p)
|
return getTimestampFromPoint(*window, p)
|
||||||
} else {
|
} else {
|
||||||
return getValueFromPoint(p.Values[idx], fType)
|
return getValueFromPoint(p.Values[idx], fType)
|
||||||
}
|
}
|
||||||
|
@ -175,7 +176,10 @@ func (self *QueryEngine) executeCountQueryWithGroupBy(query *parser.Query, yield
|
||||||
fieldTypes := map[string]*protocol.FieldDefinition_Type{}
|
fieldTypes := map[string]*protocol.FieldDefinition_Type{}
|
||||||
var inverse InverseMapper
|
var inverse InverseMapper
|
||||||
|
|
||||||
duration, ok := groupBy.GetGroupByTime()
|
duration, err := groupBy.GetGroupByTime()
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
self.coordinator.DistributeQuery(query, func(series *protocol.Series) error {
|
self.coordinator.DistributeQuery(query, func(series *protocol.Series) error {
|
||||||
var mapper Mapper
|
var mapper Mapper
|
||||||
|
@ -190,8 +194,8 @@ func (self *QueryEngine) executeCountQueryWithGroupBy(query *parser.Query, yield
|
||||||
c := counts[value]
|
c := counts[value]
|
||||||
counts[value] = c + 1
|
counts[value] = c + 1
|
||||||
|
|
||||||
if ok {
|
if duration != nil {
|
||||||
timestamps[value] = getTimestampFromPoint(duration, point)
|
timestamps[value] = getTimestampFromPoint(*duration, point)
|
||||||
} else {
|
} else {
|
||||||
timestamps[value] = point.GetTimestamp()
|
timestamps[value] = point.GetTimestamp()
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,6 +1,7 @@
|
||||||
package engine
|
package engine
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"common"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
. "launchpad.net/gocheck"
|
. "launchpad.net/gocheck"
|
||||||
|
@ -58,6 +59,15 @@ func createEngine(c *C, seriesString string) EngineI {
|
||||||
//
|
//
|
||||||
// expectedSeries must be a json array, e.g. time series must by
|
// expectedSeries must be a json array, e.g. time series must by
|
||||||
// inclosed in '[' and ']'
|
// inclosed in '[' and ']'
|
||||||
|
func runQueryRunError(engine EngineI, query string, c *C, expectedErr error) {
|
||||||
|
q, err := parser.ParseQuery(query)
|
||||||
|
c.Assert(err, IsNil)
|
||||||
|
|
||||||
|
err = engine.RunQuery(q, func(series *protocol.Series) error { return nil })
|
||||||
|
|
||||||
|
c.Assert(err, DeepEquals, expectedErr)
|
||||||
|
}
|
||||||
|
|
||||||
func runQuery(engine EngineI, query string, c *C, expectedSeries string) {
|
func runQuery(engine EngineI, query string, c *C, expectedSeries string) {
|
||||||
q, err := parser.ParseQuery(query)
|
q, err := parser.ParseQuery(query)
|
||||||
c.Assert(err, IsNil)
|
c.Assert(err, IsNil)
|
||||||
|
@ -508,6 +518,7 @@ func (self *EngineSuite) TestCountQueryWithGroupByClauseWithMultipleColumns(c *C
|
||||||
`)
|
`)
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (self *EngineSuite) TestCountQueryWithGroupByTime(c *C) {
|
func (self *EngineSuite) TestCountQueryWithGroupByTime(c *C) {
|
||||||
// make the mock coordinator return some data
|
// make the mock coordinator return some data
|
||||||
engine := createEngine(c, `
|
engine := createEngine(c, `
|
||||||
|
@ -587,3 +598,15 @@ func (self *EngineSuite) TestCountQueryWithGroupByTime(c *C) {
|
||||||
`)
|
`)
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (self *EngineSuite) TestCountQueryWithGroupByTimeInvalidNumberOfArguments(c *C) {
|
||||||
|
err := common.NewQueryError(common.WrongNumberOfArguments, "time function only accepts one argument")
|
||||||
|
engine := createEngine(c, `[]`)
|
||||||
|
runQueryRunError(engine, "select count(*) from foo group by time(1h, 1m);", c, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (self *EngineSuite) TestCountQueryWithGroupByTimeInvalidArgument(c *C) {
|
||||||
|
err := common.NewQueryError(common.InvalidArgument, "invalid argument foobar to the time function")
|
||||||
|
engine := createEngine(c, `[]`)
|
||||||
|
runQueryRunError(engine, "select count(*) from foo group by time(foobar);", c, err)
|
||||||
|
}
|
||||||
|
|
|
@ -5,6 +5,7 @@ package parser
|
||||||
import "C"
|
import "C"
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"common"
|
||||||
"fmt"
|
"fmt"
|
||||||
"time"
|
"time"
|
||||||
"unsafe"
|
"unsafe"
|
||||||
|
@ -39,17 +40,24 @@ type BoolExpression struct {
|
||||||
|
|
||||||
type GroupByClause []*Value
|
type GroupByClause []*Value
|
||||||
|
|
||||||
func (self GroupByClause) GetGroupByTime() (time.Duration, bool) {
|
func (self GroupByClause) GetGroupByTime() (*time.Duration, error) {
|
||||||
for _, groupBy := range self {
|
for _, groupBy := range self {
|
||||||
if groupBy.IsFunctionCall() {
|
if groupBy.IsFunctionCall() {
|
||||||
// TODO: check the number of arguments and return an error
|
// TODO: check the number of arguments and return an error
|
||||||
|
if len(groupBy.Elems) != 1 {
|
||||||
|
return nil, common.NewQueryError(common.WrongNumberOfArguments, "time function only accepts one argument")
|
||||||
|
}
|
||||||
// TODO: check the function name
|
// TODO: check the function name
|
||||||
// TODO: error checking
|
// TODO: error checking
|
||||||
duration, _ := time.ParseDuration(groupBy.Elems[0].Name)
|
arg := groupBy.Elems[0].Name
|
||||||
return duration, true
|
duration, err := time.ParseDuration(arg)
|
||||||
|
if err != nil {
|
||||||
|
return nil, common.NewQueryError(common.InvalidArgument, fmt.Sprintf("invalid argument %s to the time function", arg))
|
||||||
|
}
|
||||||
|
return &duration, nil
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return 0, false
|
return nil, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
type WhereCondition struct {
|
type WhereCondition struct {
|
||||||
|
|
Loading…
Reference in New Issue