fix(worker): fix between filter for non indexed predicates (hypermode…

…inc#6715)

Fixes DGRAPH-2528.
Fixes DGRAPH-2544.

This PR extends the support for the `between` filter on non-indexed predicates so that it could be used outside the root also. For example, for the given schema
```
type CarModel {
	make
	model
	year
}
model                          : string @index(term) @lang .
make                           : string @index(term) .
year                           : int .
```
The below query uses between filter on `year` predicate which is non-indexed, should be valid.
```
queryCareModel(func: type(CarModel)) @filter(between(year,2009,2010)){
				make
				model
				year
			}
```
This PR also fixes the `eq` filter for more than one argument on the non-indexed predicate which earlier used to compare with just the first of the given arguments for the `eq` filter. For eg the below query was not giving proper results earlier.
```
queryCarModel(func: type(CarModel)) @filter(eq(year,2008,2009)){
				make
				model
				year
			}
```

query/query0_test.go

@@ -3283,6 +3283,109 @@ func TestBetweenCount(t *testing.T) {
 	}
 }
 
+func TestBetweenWithIndex(t *testing.T) {
+	tests := []struct {
+		name   string
+		query  string
+		result string
+	}{
+		{
+			`Test Between on Indexed Predicate`,
+			`{
+				me(func :has(newname))  @filter(between(newname,"P1","P3")){
+					newname
+				  }
+			 }`,
+			`{"data": {"me": [{"newname": "P1"},{"newname": "P2"},{"newname": "P3"},{"newname": "P10"},{"newname": "P11"},{"newname": "P12"}]}}`,
+		},
+		{
+			`Test Between on Indexed Predicate at child Node`,
+			`{
+				me(func :has(newname))  @filter(between(newname,"P12","P2")){
+					newname
+					newfriend @filter(between(newname, "P3", "P5")){
+					  newname
+					}
+				  }
+			 }`,
+			`{"data": {"me": [{"newname": "P2", "newfriend": [{"newname": "P5"}]},{"newname": "P12"}]}}`,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			js := processQueryNoErr(t, tc.query)
+			require.JSONEq(t, js, tc.result)
+		})
+	}
+}
+func TestBetweenWithoutIndex(t *testing.T) {
+	tests := []struct {
+		name   string
+		query  string
+		result string
+	}{
+		{
+			`Test Between on Non Indexed Predicate`,
+			`
+			{
+				me(func: type(CarModel)) @filter(between(year,2009,2010)){
+					make
+					model
+					year
+				}
+			}
+			`,
+			`{"data":{"me":[{"make":"Ford","model":"Focus","year":2009},{"make":"Toyota","model":"Prius","year":2009}]}}`,
+		},
+		{
+			`Test Between filter at child node`,
+			`
+			{
+				me(func :has(newage)) @filter(between(newage,20,24)) {
+					newage
+					newfriend @filter(between(newage,25,30)){
+					  newage
+					}
+				 }
+			}
+			`,
+			`{"data": {"me": [{"newage": 21},{"newage": 22,"newfriend": [{"newage": 25},{"newage": 26}]},{"newage": 23,"newfriend": [{"newage": 27},{"newage": 28}]},{"newage": 24,"newfriend": [{"newage": 29},{"newage": 30}]}]}}`,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			js := processQueryNoErr(t, tc.query)
+			require.JSONEq(t, js, tc.result)
+		})
+	}
+
+}
+
+func TestEqFilterWithoutIndex(t *testing.T) {
+	test := struct {
+		name   string
+		query  string
+		result string
+	}{
+		`Test eq filter on Non Indexed Predicate`,
+		`
+		{
+			me(func: type(CarModel)) @filter(eq(year,2008,2009)){
+				make
+				model
+				year
+			}
+		}
+		`,
+		`{"data":{"me":[{"make":"Ford","model":"Focus","year":2008},{"make":"Ford","model":"Focus","year":2009},{"make":"Toyota","model":"Prius","year":2009}]}}`,
+	}
+
+	js := processQueryNoErr(t, test.query)
+	require.JSONEq(t, js, test.result)
+
+}
+
 var client *dgo.Dgraph
 
 func TestMain(m *testing.M) {

types/compare.go

@@ -45,3 +45,10 @@ func CompareVals(op string, arg1, arg2 Val) bool {
 	}
 	return false
 }
+
+// CompareBetween compares if the dst value lie between
+// two values val1 and val2(both inclusive).
+func CompareBetween(dst, val1, val2 Val) bool {
+	return CompareVals("ge", dst, val1) &&
+		CompareVals("le", dst, val2)
+}

worker/task.go

@@ -451,10 +451,24 @@ func (qs *queryState) handleValuePostings(ctx context.Context, args funcArgs) er
 					if val, err = types.Convert(val, srcFn.atype); err != nil {
 						return err
 					}
-					if types.CompareVals(srcFn.fname, val, srcFn.eqTokens[0]) {
-						uidList.Uids = append(uidList.Uids, q.UidList.Uids[i])
-						break
+					switch srcFn.fname {
+					case "eq":
+						for _, eqToken := range srcFn.eqTokens {
+							if types.CompareVals(srcFn.fname, val, eqToken) {
+								uidList.Uids = append(uidList.Uids, q.UidList.Uids[i])
+								break
+							}
+						}
+					case "between":
+						if types.CompareBetween(val, srcFn.eqTokens[0], srcFn.eqTokens[1]) {
+							uidList.Uids = append(uidList.Uids, q.UidList.Uids[i])
+						}
+					default:
+						if types.CompareVals(srcFn.fname, val, srcFn.eqTokens[0]) {
+							uidList.Uids = append(uidList.Uids, q.UidList.Uids[i])
+						}
 					}
+
 				} else {
 					vl.Values = append(vl.Values, newValue)
 				}
@@ -1328,8 +1342,7 @@ func (qs *queryState) handleCompareFunction(ctx context.Context, arg funcArgs) e
 		}
 	case arg.srcFn.fname == between:
 		compareFunc := func(dst types.Val) bool {
-			return types.CompareVals("ge", dst, arg.srcFn.eqTokens[0]) &&
-				types.CompareVals("le", dst, arg.srcFn.eqTokens[1])
+			return types.CompareBetween(dst, arg.srcFn.eqTokens[0], arg.srcFn.eqTokens[1])
 		}
 		if err := filterRow(0, compareFunc); err != nil {
 			return err