fix #81. Add support for IN

CHANGELOG.md

@@ -132,6 +132,7 @@
 ## Features
 
 - [Issue #80](https://github.com/influxdb/influxdb/issues/80). Support durations when specifying start and end time
+- [Issue #81](https://github.com/influxdb/influxdb/issues/81). Add support for IN
 
 ## Bugfixes
 

src/datastore/boolean_operators.go

@@ -1,29 +1,42 @@
 package datastore
 
 import (
+	"fmt"
 	"protocol"
 	"regexp"
 )
 
-type BooleanOperation func(leftValue, rightValue *protocol.FieldValue) (bool, error)
+type oldBooleanOperation func(leftValue, rightValues *protocol.FieldValue) (bool, error)
+type BooleanOperation func(leftValue *protocol.FieldValue, rightValues []*protocol.FieldValue) (bool, error)
+
+func wrapOldBooleanOperation(operation oldBooleanOperation) BooleanOperation {
+	return func(leftValue *protocol.FieldValue, rightValues []*protocol.FieldValue) (bool, error) {
+		if len(rightValues) != 1 {
+			return false, fmt.Errorf("Expected one value on the right side")
+		}
+
+		return operation(leftValue, rightValues[0])
+	}
+}
 
 var (
 	registeredOperators = map[string]BooleanOperation{}
 )
 
 func init() {
-	registeredOperators["=="] = EqualityOperator
+	registeredOperators["=="] = wrapOldBooleanOperation(EqualityOperator)
-	registeredOperators["!="] = not(EqualityOperator)
+	registeredOperators["!="] = not(wrapOldBooleanOperation(EqualityOperator))
-	registeredOperators[">="] = GreaterThanOrEqualOperator
+	registeredOperators[">="] = wrapOldBooleanOperation(GreaterThanOrEqualOperator)
-	registeredOperators[">"] = GreaterThanOperator
+	registeredOperators[">"] = wrapOldBooleanOperation(GreaterThanOperator)
-	registeredOperators["<"] = not(GreaterThanOrEqualOperator)
+	registeredOperators["<"] = not(wrapOldBooleanOperation(GreaterThanOrEqualOperator))
-	registeredOperators["<="] = not(GreaterThanOperator)
+	registeredOperators["<="] = not(wrapOldBooleanOperation(GreaterThanOperator))
-	registeredOperators["=~"] = RegexMatcherOperator
+	registeredOperators["=~"] = wrapOldBooleanOperation(RegexMatcherOperator)
-	registeredOperators["!~"] = not(RegexMatcherOperator)
+	registeredOperators["!~"] = not(wrapOldBooleanOperation(RegexMatcherOperator))
+	registeredOperators["in"] = InOperator
 }
 
 func not(op BooleanOperation) BooleanOperation {
-	return func(leftValue, rightValue *protocol.FieldValue) (bool, error) {
+	return func(leftValue *protocol.FieldValue, rightValue []*protocol.FieldValue) (bool, error) {
 		ok, err := op(leftValue, rightValue)
 		return !ok, err
 	}
@@ -135,3 +148,30 @@ func GreaterThanOperator(leftValue, rightValue *protocol.FieldValue) (bool, erro
 		return false, nil
 	}
 }
+
+func InOperator(leftValue *protocol.FieldValue, rightValue []*protocol.FieldValue) (bool, error) {
+	for _, v := range rightValue {
+		v1, v2, cType := coerceValues(leftValue, v)
+
+		var result bool
+
+		switch cType {
+		case TYPE_STRING:
+			result = v1.(string) == v2.(string)
+		case TYPE_INT:
+			result = v1.(int64) == v2.(int64)
+		case TYPE_DOUBLE:
+			result = v1.(float64) == v2.(float64)
+		case TYPE_BOOL:
+			result = v1.(bool) == v2.(bool)
+		default:
+			result = false
+		}
+
+		if result {
+			return true, nil
+		}
+	}
+
+	return false, nil
+}

src/datastore/filtering.go

@@ -7,20 +7,27 @@ import (
 	"strconv"
 )
 
-func getExpressionValue(expr *parser.Expression, fields []string, point *protocol.Point) (*protocol.FieldValue, error) {
+func getExpressionValue(expr *parser.Expression, fields []string, point *protocol.Point) ([]*protocol.FieldValue, error) {
+
+	values, _ := expr.GetLeftValues()
 
 	if value, ok := expr.GetLeftValue(); ok {
+		values = []*parser.Value{value}
+	}
+
+	fieldValues := []*protocol.FieldValue{}
+	for _, value := range values {
 		switch value.Type {
 		case parser.ValueFunctionCall:
 			return nil, fmt.Errorf("Cannot process function call %s in expression", value.Name)
 		case parser.ValueFloat:
 			value, _ := strconv.ParseFloat(value.Name, 64)
-			return &protocol.FieldValue{DoubleValue: &value}, nil
+			fieldValues = append(fieldValues, &protocol.FieldValue{DoubleValue: &value})
 		case parser.ValueInt:
 			value, _ := strconv.ParseInt(value.Name, 10, 64)
-			return &protocol.FieldValue{Int64Value: &value}, nil
+			fieldValues = append(fieldValues, &protocol.FieldValue{Int64Value: &value})
 		case parser.ValueString, parser.ValueRegex:
-			return &protocol.FieldValue{StringValue: &value.Name}, nil
+			fieldValues = append(fieldValues, &protocol.FieldValue{StringValue: &value.Name})
 		case parser.ValueTableName, parser.ValueSimpleName:
 
 			// TODO: optimize this so we don't have to lookup the column everytime
@@ -36,11 +43,13 @@ func getExpressionValue(expr *parser.Expression, fields []string, point *protoco
 				return nil, fmt.Errorf("Cannot find column %s", value.Name)
 			}
 
-			return point.Values[fieldIdx], nil
+			fieldValues = append(fieldValues, point.Values[fieldIdx])
+		default:
+			return nil, fmt.Errorf("Cannot evaluate expression")
 		}
 	}
 
-	return nil, fmt.Errorf("Cannot evaluate expression")
+	return fieldValues, nil
 }
 
 func matchesExpression(expr *parser.BoolExpression, fields []string, point *protocol.Point) (bool, error) {
@@ -54,7 +63,7 @@ func matchesExpression(expr *parser.BoolExpression, fields []string, point *prot
 	}
 
 	operator := registeredOperators[expr.Operation]
-	return operator(leftValue, rightValue)
+	return operator(leftValue[0], rightValue)
 }
 
 func matches(condition *parser.WhereCondition, fields []string, point *protocol.Point) (bool, error) {

src/datastore/filtering_test.go

@@ -10,6 +10,35 @@ type FilteringSuite struct{}
 
 var _ = Suite(&FilteringSuite{})
 
+func (self *FilteringSuite) TestInOperatorFiltering(c *C) {
+	queryStr := "select * from t where column_one in (100, 85);"
+	query, err := parser.ParseQuery(queryStr)
+	c.Assert(err, IsNil)
+
+	series, err := common.StringToSeriesArray(`
+[
+ {
+   "points": [
+     {"values": [{"int64_value": 100},{"int64_value": 5 }], "timestamp": 1381346631, "sequence_number": 1},
+     {"values": [{"int64_value": 85},{"int64_value": 6 }], "timestamp": 1381346631, "sequence_number": 1},
+     {"values": [{"int64_value": 90 },{"int64_value": 15}], "timestamp": 1381346632, "sequence_number": 1}
+   ],
+   "name": "t",
+   "fields": ["column_one", "column_two"]
+ }
+]
+`)
+	c.Assert(err, IsNil)
+	result, err := Filter(query, series[0])
+	c.Assert(err, IsNil)
+	c.Assert(result, NotNil)
+	c.Assert(result.Points, HasLen, 2)
+	c.Assert(*result.Points[0].Values[0].Int64Value, Equals, int64(100))
+	c.Assert(*result.Points[0].Values[1].Int64Value, Equals, int64(5))
+	c.Assert(*result.Points[1].Values[0].Int64Value, Equals, int64(85))
+	c.Assert(*result.Points[1].Values[1].Int64Value, Equals, int64(6))
+}
+
 func (self *FilteringSuite) TestEqualityFiltering(c *C) {
 	queryStr := "select * from t where column_one == 100 and column_two != 6;"
 	query, err := parser.ParseQuery(queryStr)

src/integration/benchmark_test.go

@@ -305,6 +305,31 @@ func (self *IntegrationSuite) TestFilterWithLimit(c *C) {
 	c.Assert(data[0].Points, HasLen, 1)
 }
 
+// issue #81
+func (self *IntegrationSuite) TestFilterWithInClause(c *C) {
+	for i := 0; i < 3; i++ {
+		err := self.server.WriteData(fmt.Sprintf(`
+[
+  {
+     "name": "test_in_clause",
+     "columns": ["cpu", "host"],
+     "points": [[%d, "hosta"], [%d, "hostb"]]
+  }
+]
+`, 60+i*10, 70+i*10))
+		c.Assert(err, IsNil)
+		time.Sleep(1 * time.Second)
+	}
+	bs, err := self.server.RunQuery("select host, cpu from test_in_clause where host in ('hostb') order asc limit 1")
+	c.Assert(err, IsNil)
+	data := []*h.SerializedSeries{}
+	err = json.Unmarshal(bs, &data)
+	c.Assert(data, HasLen, 1)
+	c.Assert(data[0].Name, Equals, "test_in_clause")
+	c.Assert(data[0].Columns, HasLen, 4)
+	c.Assert(data[0].Points, HasLen, 1)
+}
+
 // issue #36
 func (self *IntegrationSuite) TestInnerJoin(c *C) {
 	for i := 0; i < 3; i++ {

src/parser/frees.c

@@ -56,6 +56,8 @@ free_expression(expression *expr)
 {
   if (expr->op == 0) {
     free_value((value*)expr->left);
+  } else if (expr->op == 1) {
+    free_value_array((value_array*)expr->left);
   } else {
     free_expression((expression*) expr->left);
     free_expression(expr->right);

src/parser/parser.go

@@ -165,8 +165,15 @@ func (self *Expression) GetLeftValue() (*Value, bool) {
 	return nil, false
 }
 
+func (self *Expression) GetLeftValues() ([]*Value, bool) {
+	if self.Operation == 1 {
+		return self.Left.([]*Value), true
+	}
+	return nil, false
+}
+
 func (self *Expression) GetLeftExpression() (*Expression, bool) {
-	if self.Operation != 0 {
+	if self.Operation > 1 {
 		return self.Left.(*Expression), true
 	}
 	return nil, false
@@ -277,6 +284,14 @@ func GetExpression(expr *C.expression) (*Expression, error) {
 		expression.Left = value
 		expression.Operation = byte(expr.op)
 		expression.Right = nil
+	} else if expr.op == 1 {
+		value, err := GetValueArray((*C.value_array)(expr.left))
+		if err != nil {
+			return nil, err
+		}
+		expression.Left = value
+		expression.Operation = byte(expr.op)
+		expression.Right = nil
 	} else {
 		var err error
 		expression.Left, err = GetExpression((*C.expression)(expr.left))

src/parser/parser_test.go

@@ -431,6 +431,23 @@ func (self *QueryParserSuite) TestTimeConditionWithFloats(c *C) {
 	}
 }
 
+func (self *QueryParserSuite) TestQureyWithInCondition(c *C) {
+	query := "select * from foo where bar in ('baz', 'bazz')"
+	q, err := ParseQuery(query)
+	c.Assert(err, IsNil)
+	condition := q.GetWhereCondition()
+	expr, ok := condition.GetBoolExpression()
+	c.Assert(ok, Equals, true)
+	left, _ := expr.Left.GetLeftValue()
+	c.Assert(expr.Operation, Equals, "in")
+	right, ok := expr.Right.GetLeftValues()
+	c.Assert(ok, Equals, true)
+	c.Assert(left.Name, Equals, "bar")
+	c.Assert(right, HasLen, 2)
+	c.Assert(right[0].Name, Equals, "baz")
+	c.Assert(right[1].Name, Equals, "bazz")
+}
+
 // TODO:
 // insert into user.events.count.per_day select count(*) from user.events where time<forever group by time(1d)
 // insert into :series_name.percentiles.95 select percentile(95,value) from stats.* where time<forever group by time(1d)

src/parser/query.lex

@@ -33,6 +33,7 @@ static int yycolumn = 1;
 "limit"                   { return LIMIT; }
 "order"                   { return ORDER; }
 "asc"                     { return ASC; }
+"in"                      { yylval->string = strdup(yytext); return OPERATION_IN; }
 "desc"                    { return DESC; }
 "group"                   { return GROUP; }
 "by"                      { return BY; }
@@ -66,7 +67,7 @@ static int yycolumn = 1;
 
 [a-zA-Z][a-zA-Z0-9._-]*   { yylval->string = strdup(yytext); return TABLE_NAME; }
 
-\'.*\'                    {
+\'[^\']*\'                    {
   yytext[yyleng-1] = '\0';
   yylval->string = strdup(yytext+1);
   return STRING_VALUE;

src/parser/query.yacc

@@ -66,7 +66,7 @@ value *create_value(char *name, int type, char is_case_insensitive, value_array
 // define the precedence of these operators
 %left  OR
 %left  AND
-%nonassoc <string> OPERATION_EQUAL OPERATION_NE OPERATION_GT OPERATION_LT OPERATION_LE OPERATION_GE
+%nonassoc <string> OPERATION_EQUAL OPERATION_NE OPERATION_GT OPERATION_LT OPERATION_LE OPERATION_GE OPERATION_IN
 %left  <character> '+' '-'
 %left  <character> '*' '/'
 
@@ -408,6 +408,17 @@ BOOL_EXPRESSION:
           $$->right = $3;
         }
         |
+        EXPRESSION OPERATION_IN '(' VALUES ')'
+        {
+          $$ = malloc(sizeof(bool_expression));
+          $$->left = $1;
+          $$->op = $2;
+          $$->right = malloc(sizeof(expression));
+          $$->right->left = $4;
+          $$->right->op = '\1';
+          $$->right->right = NULL;
+        }
+        |
         EXPRESSION REGEX_OP REGEX_VALUE
         {
           $$ = malloc(sizeof(bool_expression));

src/parser/query_api.go

@@ -262,8 +262,18 @@ func getReferencedColumnsFromExpression(expr *Expression, mapping map[string][]s
 		return
 	}
 
-	value, _ := expr.GetLeftValue()
+	values, ok := expr.GetLeftValues()
-	notAssigned = append(notAssigned, getReferencedColumnsFromValue(value, mapping)...)
+	if !ok {
+		value, ok := expr.GetLeftValue()
+		if ok {
+			values = []*Value{value}
+		}
+	}
+
+	for _, v := range values {
+		notAssigned = append(notAssigned, getReferencedColumnsFromValue(v, mapping)...)
+	}
+
 	return
 }
 

src/parser/query_api_test.go

@@ -65,6 +65,18 @@ func (self *QueryApiSuite) TestGetReferencedColumns(c *C) {
 	}
 }
 
+func (self *QueryApiSuite) TestGetReferencedColumnsWithInClause(c *C) {
+	queryStr := "select value1, sum(value2) from t where value In (90.0, 100.0) group by value3;"
+	query, err := ParseQuery(queryStr)
+	c.Assert(err, IsNil)
+	columns := query.GetReferencedColumns()
+	c.Assert(columns, HasLen, 1)
+	for v, columns := range columns {
+		c.Assert(columns, DeepEquals, []string{"value", "value1", "value2", "value3"})
+		c.Assert(v.Name, Equals, "t")
+	}
+}
+
 func (self *QueryApiSuite) TestGetReferencedColumnsReturnsTheStarAsAColumn(c *C) {
 	queryStr := "select * from events;"
 	query, err := ParseQuery(queryStr)