Merge pull request #1210 from influxdb/planner

Add constant folding.

influxql/ast.go

@@ -537,7 +537,9 @@ type TimeLiteral struct {
 }
 
 // String returns a string representation of the literal.
-func (l *TimeLiteral) String() string { return l.Val.UTC().Format("2006-01-02 15:04:05.999") }
+func (l *TimeLiteral) String() string {
+	return `"` + l.Val.UTC().Format("2006-01-02 15:04:05.999999") + `"`
+}
 
 // DurationLiteral represents a duration literal.
 type DurationLiteral struct {
@@ -573,6 +575,221 @@ type Wildcard struct{}
 // String returns a string representation of the wildcard.
 func (e *Wildcard) String() string { return "*" }
 
+// Fold performs constant folding on an expression.
+// The function, "now()", is expanded into the current time during folding.
+func Fold(expr Expr, now *time.Time) Expr {
+	switch expr := expr.(type) {
+	case *Call:
+		// Replace "now()" with current time.
+		if strings.ToLower(expr.Name) == "now" && now != nil {
+			return &TimeLiteral{Val: *now}
+		}
+
+		// Fold call arguments.
+		for i, arg := range expr.Args {
+			expr.Args[i] = Fold(arg, now)
+		}
+		return expr
+
+	case *BinaryExpr:
+		// Fold and evaluate binary expression.
+		return foldBinaryExpr(expr, now)
+
+	case *ParenExpr:
+		// Fold inside expression.
+		// Return inside expression if not a binary expression.
+		expr.Expr = Fold(expr.Expr, now)
+		if _, ok := expr.Expr.(*BinaryExpr); !ok {
+			return expr.Expr
+		}
+		return expr
+
+	default:
+		return expr
+	}
+}
+
+// foldBinaryExpr performs constant folding if the binary expression has two literals.
+func foldBinaryExpr(expr *BinaryExpr, now *time.Time) Expr {
+	// Fold both sides of binary expression first.
+	expr.LHS = Fold(expr.LHS, now)
+	expr.RHS = Fold(expr.RHS, now)
+
+	// Evaluate operations if both sides are the same type.
+	switch lhs := expr.LHS.(type) {
+	case *NumberLiteral:
+		if _, ok := expr.RHS.(*NumberLiteral); ok {
+			return foldNumberLiterals(expr)
+		}
+	case *BooleanLiteral:
+		if _, ok := expr.RHS.(*BooleanLiteral); ok {
+			return foldBooleanLiterals(expr)
+		}
+	case *TimeLiteral:
+		switch expr.RHS.(type) {
+		case *TimeLiteral:
+			return foldTimeLiterals(expr)
+		case *DurationLiteral:
+			return foldTimeDurationLiterals(expr)
+		}
+	case *DurationLiteral:
+		switch rhs := expr.RHS.(type) {
+		case *DurationLiteral:
+			return foldDurationLiterals(expr)
+		case *NumberLiteral:
+			return foldDurationNumberLiterals(expr)
+		case *TimeLiteral:
+			if expr.Op == ADD {
+				return &TimeLiteral{Val: rhs.Val.Add(lhs.Val)}
+			}
+		}
+	case *StringLiteral:
+		if rhs, ok := expr.RHS.(*StringLiteral); ok && expr.Op == ADD {
+			return &StringLiteral{Val: lhs.Val + rhs.Val}
+		}
+	}
+
+	return expr
+}
+
+// foldNumberLiterals performs constant folding on two number literals.
+func foldNumberLiterals(expr *BinaryExpr) Expr {
+	lhs := expr.LHS.(*NumberLiteral)
+	rhs := expr.RHS.(*NumberLiteral)
+
+	switch expr.Op {
+	case ADD:
+		return &NumberLiteral{Val: lhs.Val + rhs.Val}
+	case SUB:
+		return &NumberLiteral{Val: lhs.Val - rhs.Val}
+	case MUL:
+		return &NumberLiteral{Val: lhs.Val * rhs.Val}
+	case DIV:
+		if rhs.Val == 0 {
+			return &NumberLiteral{Val: 0}
+		}
+		return &NumberLiteral{Val: lhs.Val / rhs.Val}
+	case EQ:
+		return &BooleanLiteral{Val: lhs.Val == rhs.Val}
+	case NEQ:
+		return &BooleanLiteral{Val: lhs.Val != rhs.Val}
+	case GT:
+		return &BooleanLiteral{Val: lhs.Val > rhs.Val}
+	case GTE:
+		return &BooleanLiteral{Val: lhs.Val >= rhs.Val}
+	case LT:
+		return &BooleanLiteral{Val: lhs.Val < rhs.Val}
+	case LTE:
+		return &BooleanLiteral{Val: lhs.Val <= rhs.Val}
+	default:
+		return expr
+	}
+}
+
+// foldBooleanLiterals performs constant folding on two boolean literals.
+func foldBooleanLiterals(expr *BinaryExpr) Expr {
+	lhs := expr.LHS.(*BooleanLiteral)
+	rhs := expr.RHS.(*BooleanLiteral)
+
+	switch expr.Op {
+	case EQ:
+		return &BooleanLiteral{Val: lhs.Val == rhs.Val}
+	case NEQ:
+		return &BooleanLiteral{Val: lhs.Val != rhs.Val}
+	case AND:
+		return &BooleanLiteral{Val: lhs.Val && rhs.Val}
+	case OR:
+		return &BooleanLiteral{Val: lhs.Val || rhs.Val}
+	default:
+		return expr
+	}
+}
+
+// foldTimeLiterals performs constant folding on two time literals.
+func foldTimeLiterals(expr *BinaryExpr) Expr {
+	lhs := expr.LHS.(*TimeLiteral)
+	rhs := expr.RHS.(*TimeLiteral)
+
+	switch expr.Op {
+	case SUB:
+		return &DurationLiteral{Val: lhs.Val.Sub(rhs.Val)}
+	case EQ:
+		return &BooleanLiteral{Val: lhs.Val.Equal(rhs.Val)}
+	case NEQ:
+		return &BooleanLiteral{Val: !lhs.Val.Equal(rhs.Val)}
+	case GT:
+		return &BooleanLiteral{Val: lhs.Val.After(rhs.Val)}
+	case GTE:
+		return &BooleanLiteral{Val: lhs.Val.After(rhs.Val) || lhs.Val.Equal(rhs.Val)}
+	case LT:
+		return &BooleanLiteral{Val: lhs.Val.Before(rhs.Val)}
+	case LTE:
+		return &BooleanLiteral{Val: lhs.Val.Before(rhs.Val) || lhs.Val.Equal(rhs.Val)}
+	default:
+		return expr
+	}
+}
+
+// foldTimeDurationLiterals performs constant folding on a time and duration literal.
+func foldTimeDurationLiterals(expr *BinaryExpr) Expr {
+	lhs := expr.LHS.(*TimeLiteral)
+	rhs := expr.RHS.(*DurationLiteral)
+
+	switch expr.Op {
+	case ADD:
+		return &TimeLiteral{Val: lhs.Val.Add(rhs.Val)}
+	case SUB:
+		return &TimeLiteral{Val: lhs.Val.Add(-rhs.Val)}
+	default:
+		return expr
+	}
+}
+
+// foldDurationLiterals performs constant folding on two duration literals.
+func foldDurationLiterals(expr *BinaryExpr) Expr {
+	lhs := expr.LHS.(*DurationLiteral)
+	rhs := expr.RHS.(*DurationLiteral)
+
+	switch expr.Op {
+	case ADD:
+		return &DurationLiteral{Val: lhs.Val + rhs.Val}
+	case SUB:
+		return &DurationLiteral{Val: lhs.Val - rhs.Val}
+	case EQ:
+		return &BooleanLiteral{Val: lhs.Val == rhs.Val}
+	case NEQ:
+		return &BooleanLiteral{Val: lhs.Val != rhs.Val}
+	case GT:
+		return &BooleanLiteral{Val: lhs.Val > rhs.Val}
+	case GTE:
+		return &BooleanLiteral{Val: lhs.Val >= rhs.Val}
+	case LT:
+		return &BooleanLiteral{Val: lhs.Val < rhs.Val}
+	case LTE:
+		return &BooleanLiteral{Val: lhs.Val <= rhs.Val}
+	default:
+		return expr
+	}
+}
+
+// foldDurationNumberLiterals performs constant folding on duration and number literal.
+func foldDurationNumberLiterals(expr *BinaryExpr) Expr {
+	lhs := expr.LHS.(*DurationLiteral)
+	rhs := expr.RHS.(*NumberLiteral)
+
+	switch expr.Op {
+	case MUL:
+		return &DurationLiteral{Val: lhs.Val * time.Duration(rhs.Val)}
+	case DIV:
+		if rhs.Val == 0 {
+			return &DurationLiteral{Val: 0}
+		}
+		return &DurationLiteral{Val: lhs.Val / time.Duration(rhs.Val)}
+	default:
+		return expr
+	}
+}
+
 // Visitor can be called by Walk to traverse an AST hierarchy.
 // The Visit() function is called once per node.
 type Visitor interface {

influxql/ast_test.go

@@ -85,3 +85,78 @@ func TestSelectStatement_Substatement(t *testing.T) {
 		}
 	}
 }
+
+// Ensure an expression can be folded.
+func TestFold(t *testing.T) {
+	for i, tt := range []struct {
+		in  string
+		out string
+	}{
+		// Number literals.
+		{`1 + 2`, `3.000`},
+		{`(foo*2) + (4/2) + (3 * 5) - 0.5`, `(foo * 2.000) + 16.500`},
+		{`foo(bar(2 + 3), 4)`, `foo(bar(5.000), 4.000)`},
+		{`4 / 0`, `0.000`},
+		{`4 = 4`, `true`},
+		{`4 != 4`, `false`},
+		{`6 > 4`, `true`},
+		{`4 >= 4`, `true`},
+		{`4 < 6`, `true`},
+		{`4 <= 4`, `true`},
+		{`4 AND 5`, `4.000 AND 5.000`},
+
+		// Boolean literals.
+		{`true AND false`, `false`},
+		{`true OR false`, `true`},
+		{`true = false`, `false`},
+		{`true != false`, `true`},
+		{`true + false`, `true + false`},
+
+		// Time literals.
+		{`now() + 2h`, `"2000-01-01 02:00:00"`},
+		{`now() / 2h`, `"2000-01-01 00:00:00" / 2h`},
+		{`4µ + now()`, `"2000-01-01 00:00:00.000004"`},
+		{`now() = now()`, `true`},
+		{`now() != now()`, `false`},
+		{`now() < now() + 1h`, `true`},
+		{`now() <= now() + 1h`, `true`},
+		{`now() >= now() - 1h`, `true`},
+		{`now() > now() - 1h`, `true`},
+		{`now() - (now() - 60s)`, `1m`},
+		{`now() AND now()`, `"2000-01-01 00:00:00" AND "2000-01-01 00:00:00"`},
+
+		// Duration literals.
+		{`10m + 1h - 60s`, `69m`},
+		{`(10m / 2) * 5`, `25m`},
+		{`60s = 1m`, `true`},
+		{`60s != 1m`, `false`},
+		{`60s < 1h`, `true`},
+		{`60s <= 1h`, `true`},
+		{`60s > 12s`, `true`},
+		{`60s >= 1m`, `true`},
+		{`60s AND 1m`, `1m AND 1m`},
+		{`60m / 0`, `0s`},
+		{`60m + 50`, `1h + 50.000`},
+
+		// String literals.
+		{`"foo" + 'bar'`, `"foobar"`},
+	} {
+		// Fold expression.
+		now := mustParseTime("2000-01-01T00:00:00Z")
+		expr := influxql.Fold(MustParseExpr(tt.in), &now)
+
+		// Compare with expected output.
+		if out := expr.String(); tt.out != out {
+			t.Errorf("%d. %s: unexpected expr:\n\nexp=%s\n\ngot=%s\n\n", i, tt.in, tt.out, out)
+			continue
+		}
+	}
+}
+
+// Ensure an a "now()" call is not folded when now is not passed in.
+func TestFold_WithoutNow(t *testing.T) {
+	expr := influxql.Fold(MustParseExpr(`now()`), nil)
+	if s := expr.String(); s != `now()` {
+		t.Fatalf("unexpected expr: %s", s)
+	}
+}

influxql/engine.go

@@ -118,6 +118,15 @@ type DB interface {
 // Planner represents an object for creating execution plans.
 type Planner struct {
 	DB  DB
+	Now func() time.Time
+}
+
+// NewPlanner returns a new instance of Planner.
+func NewPlanner(db DB) *Planner {
+	return &Planner{
+		DB:  db,
+		Now: time.Now,
+	}
 }
 
 func (p *Planner) Plan(stmt *SelectStatement) (*Executor, error) {
@@ -192,7 +201,7 @@ func (p *Planner) planCall(e *Executor, c *Call) (processor, error) {
 	tags := make(map[string]string) // TODO: Extract tags.
 
 	// Find field.
-	fname := strings.TrimPrefix(ref.Val, name)
+	fname := strings.TrimPrefix(ref.Val, name+".")
 	fieldID, typ := e.db.Field(name, fname)
 	if fieldID == 0 {
 		return nil, fmt.Errorf("field not found: %s.%s", name, fname)
@@ -257,14 +266,21 @@ func (e *Executor) Execute() (<-chan *Row, error) {
 func (e *Executor) execute(out chan *Row) {
 	// TODO: Support multi-value rows.
 
-	// Combine values from each processor.
 	row := &Row{}
-	row.Values = []map[string]interface{}{
-		make(map[string]interface{}),
-	}
-	for i, p := range e.processors {
-		f := e.stmt.Fields[i]
 
+	// Create column names.
+	row.Columns = make([]string, len(e.stmt.Fields))
+	for i, f := range e.stmt.Fields {
+		name := f.Name()
+		if name == "" {
+			name = fmt.Sprintf("col%d", i)
+		}
+		row.Columns[i] = name
+	}
+
+	// Combine values from each processor.
+	row.Values = [][]interface{}{make([]interface{}, 1, len(e.processors))}
+	for i, p := range e.processors {
 		// Retrieve data from the processor.
 		m, ok := <-p.C()
 		if !ok {
@@ -273,11 +289,8 @@ func (e *Executor) execute(out chan *Row) {
 
 		// Set values on returned row.
 		row.Name = p.name()
-		for k, v := range m {
+		for _, v := range m {
-			if k != 0 {
+			row.Values[0][i] = v
-				row.Values[0]["timestamp"] = time.Unix(0, k).UTC().Format(time.RFC3339Nano)
-			}
-			row.Values[0][f.Name()] = v
 		}
 	}
 
@@ -466,7 +479,11 @@ type reduceFunc func(int64, []interface{}, *reducer)
 
 // reduceCount computes the number of values for each key.
 func reduceCount(key int64, values []interface{}, r *reducer) {
-	r.emit(key, len(values))
+	var n float64
+	for _, v := range values {
+		n += v.(float64)
+	}
+	r.emit(key, n)
 }
 
 // processor represents an object for joining reducer output.
@@ -540,7 +557,8 @@ type Iterator interface {
 type Row struct {
 	Name    string            `json:"name,omitempty"`
 	Tags    map[string]string `json:"tags,omitempty"`
-	Values []map[string]interface{} `json:"values,omitempty"`
+	Columns []string          `json:"columns"`
+	Values  [][]interface{}   `json:"values,omitempty"`
 	Err     error             `json:"err,omitempty"`
 }
 

influxql/engine_test.go

@@ -12,9 +12,15 @@ import (
 	"github.com/influxdb/influxdb/influxql"
 )
 
+// Ensure the planner can plan and execute a query.
 func TestPlanner_Plan(t *testing.T) {
 	db := NewDB()
 	db.WriteSeries("cpu", map[string]string{}, "2000-01-01T00:00:00Z", map[string]interface{}{"value": float64(100)})
+	db.WriteSeries("cpu", map[string]string{}, "2000-01-01T00:00:10Z", map[string]interface{}{"value": float64(90)})
+	db.WriteSeries("cpu", map[string]string{}, "2000-01-01T00:00:20Z", map[string]interface{}{"value": float64(80)})
+	db.WriteSeries("cpu", map[string]string{}, "2000-01-01T00:00:30Z", map[string]interface{}{"value": float64(70)})
+	db.WriteSeries("cpu", map[string]string{}, "2000-01-01T00:00:40Z", map[string]interface{}{"value": float64(60)})
+	db.WriteSeries("cpu", map[string]string{}, "2000-01-01T00:00:50Z", map[string]interface{}{"value": float64(50)})
 
 	for i, tt := range []struct {
 		q  string          // querystring
@@ -27,15 +33,16 @@ func TestPlanner_Plan(t *testing.T) {
 				{
 					Name:    "cpu",
 					Tags:    map[string]string{},
-					Values: []map[string]interface{}{
+					Columns: []string{"count"},
-						{"count": 1},
+					Values: [][]interface{}{
+						{6},
 					},
 				},
 			},
 		},
 	} {
 		// Plan statement.
-		var p = influxql.Planner{DB: db}
+		var p = influxql.NewPlanner(db)
 		e, err := p.Plan(MustParseSelectStatement(tt.q))
 		if err != nil {
 			t.Errorf("%d. %s: plan error: %s", i, tt.q, err)
@@ -57,7 +64,7 @@ func TestPlanner_Plan(t *testing.T) {
 
 		// Compare resultset.
 		if b0, b1 := mustMarshalJSON(tt.rs), mustMarshalJSON(rs); string(b0) != string(b1) {
-			t.Errorf("%d. %s: resultset mismatch:\n\nexp=%s\n\ngot=%s\n\n", i, tt.q, b0, b1)
+			t.Errorf("%d. resultset mismatch:\n\n%s\n\nexp=%s\n\ngot=%s\n\n", i, tt.q, b0, b1)
 			continue
 		}
 	}
@@ -208,7 +215,7 @@ type iterator struct {
 }
 
 // Next returns the next point's timestamp and field value.
-func (i *iterator) Next() (int64, interface{}) {
+func (i *iterator) Next() (timestamp int64, value interface{}) {
 	for {
 		// If index is beyond points range then return nil.
 		if i.index > len(i.points)-1 {

influxql/parser.go

@@ -677,7 +677,9 @@ func ParseDuration(s string) (time.Duration, error) {
 
 // FormatDuration formats a duration to a string.
 func FormatDuration(d time.Duration) string {
-	if d%(7*24*time.Hour) == 0 {
+	if d == 0 {
+		return "0s"
+	} else if d%(7*24*time.Hour) == 0 {
 		return fmt.Sprintf("%dw", d/(7*24*time.Hour))
 	} else if d%(24*time.Hour) == 0 {
 		return fmt.Sprintf("%dd", d/(24*time.Hour))

influxql/parser_test.go

@@ -451,6 +451,15 @@ func MustParseSelectStatement(s string) *influxql.SelectStatement {
 	return stmt.(*influxql.SelectStatement)
 }
 
+// MustParseExpr parses an expression. Panic on error.
+func MustParseExpr(s string) influxql.Expr {
+	expr, err := influxql.NewParser(strings.NewReader(s)).ParseExpr()
+	if err != nil {
+		panic(err.Error())
+	}
+	return expr
+}
+
 // errstring converts an error to its string representation.
 func errstring(err error) string {
 	if err != nil {