Repository: genjidb/genji
Pull request #367 (2021-03-07 13:28:00)

Use PkScan operator
This PR modifies the planner to use the stream.PkScan operator when possible.
This operator behaves like an index but is faster and space efficient because it uses
the table tree rather than a separate index tree.

Benchmarks:

Before

```bash
$ go test -count=5 -benchmem -run=^$ -bench BenchmarkSelectPk | tee old.txt
goos: linux
goarch: amd64
pkg: github.com/genjidb/genji
cpu: Intel(R) Core(TM) i7-8700K CPU @ 3.70GHz
BenchmarkSelectPk/00001-12                 60174             17452 ns/op            4184 B/op         90 allocs/op
BenchmarkSelectPk/00001-12                 67738             17082 ns/op            4183 B/op         90 allocs/op
BenchmarkSelectPk/00001-12                 71444             16640 ns/op            4183 B/op         90 allocs/op
BenchmarkSelectPk/00001-12                 69579             17069 ns/op            4183 B/op         90 allocs/op
BenchmarkSelectPk/00001-12                 67123             16799 ns/op            4183 B/op         90 allocs/op
BenchmarkSelectPk/00010-12                 63070             18711 ns/op            5224 B/op        135 allocs/op
BenchmarkSelectPk/00010-12                 62373             19149 ns/op            5224 B/op        135 allocs/op
BenchmarkSelectPk/00010-12                 61804             18993 ns/op            5223 B/op        135 allocs/op
BenchmarkSelectPk/00010-12                 61906             19465 ns/op            5226 B/op        135 allocs/op
BenchmarkSelectPk/00010-12                 61243             18744 ns/op            5225 B/op        135 allocs/op
BenchmarkSelectPk/00100-12                 12324             96532 ns/op           15691 B/op        585 allocs/op
BenchmarkSelectPk/00100-12                 12452             95986 ns/op           15687 B/op        585 allocs/op
BenchmarkSelectPk/00100-12                 12182             97492 ns/op           15686 B/op        585 allocs/op
BenchmarkSelectPk/00100-12                 12422             96658 ns/op           15689 B/op        585 allocs/op
BenchmarkSelectPk/00100-12                 12366             98609 ns/op           15685 B/op        585 allocs/op
BenchmarkSelectPk/01000-12                  1333            880389 ns/op          138400 B/op       6830 allocs/op
BenchmarkSelectPk/01000-12                  1354            908564 ns/op          138278 B/op       6830 allocs/op
BenchmarkSelectPk/01000-12                  1345            879852 ns/op          138388 B/op       6830 allocs/op
BenchmarkSelectPk/01000-12                  1336            890207 ns/op          138356 B/op       6830 allocs/op
BenchmarkSelectPk/01000-12                  1342            882389 ns/op          138293 B/op       6830 allocs/op
BenchmarkSelectPk/10000-12                   132           9377784 ns/op         1371696 B/op      69839 allocs/op
BenchmarkSelectPk/10000-12                   129           9044030 ns/op         1370267 B/op      69838 allocs/op
BenchmarkSelectPk/10000-12                   132           9024456 ns/op         1371280 B/op      69839 allocs/op
BenchmarkSelectPk/10000-12                   127           9021765 ns/op         1370703 B/op      69838 allocs/op
BenchmarkSelectPk/10000-12                   132           9436749 ns/op         1370667 B/op      69838 allocs/op
```

After

```bash
$ go test -count=5 -benchmem -run=^$ -bench BenchmarkSelectPk | tee new.txt
goos: linux
goarch: amd64
pkg: github.com/genjidb/genji
cpu: Intel(R) Core(TM) i7-8700K CPU @ 3.70GHz
BenchmarkSelectPk/00001-12                 59692             18439 ns/op            4632 B/op         95 allocs/op
BenchmarkSelectPk/00001-12                 67766             17365 ns/op            4632 B/op         95 allocs/op
BenchmarkSelectPk/00001-12                 68274             17324 ns/op            4632 B/op         95 allocs/op
BenchmarkSelectPk/00001-12                 68650             17540 ns/op            4632 B/op         95 allocs/op
BenchmarkSelectPk/00001-12                 67000             17293 ns/op            4632 B/op         95 allocs/op
BenchmarkSelectPk/00010-12                 68598             17529 ns/op            4632 B/op         95 allocs/op
BenchmarkSelectPk/00010-12                 65353             17302 ns/op            4632 B/op         95 allocs/op
BenchmarkSelectPk/00010-12                 69853             17606 ns/op            4632 B/op         95 allocs/op
BenchmarkSelectPk/00010-12                 67478             17089 ns/op            4632 B/op         95 allocs/op
BenchmarkSelectPk/00010-12                 62709             17468 ns/op            4632 B/op         95 allocs/op
BenchmarkSelectPk/00100-12                 53097             22570 ns/op            4632 B/op         95 allocs/op
BenchmarkSelectPk/00100-12                 55345             22118 ns/op            4632 B/op         95 allocs/op
BenchmarkSelectPk/00100-12                 54732             22356 ns/op            4632 B/op         95 allocs/op
BenchmarkSelectPk/00100-12                 54140             21778 ns/op            4632 B/op         95 allocs/op
BenchmarkSelectPk/00100-12                 53340             22305 ns/op            4632 B/op         95 allocs/op
BenchmarkSelectPk/01000-12                 54985             22117 ns/op            4648 B/op         97 allocs/op
BenchmarkSelectPk/01000-12                 54417             21713 ns/op            4648 B/op         97 allocs/op
BenchmarkSelectPk/01000-12                 55604             22208 ns/op            4648 B/op         97 allocs/op
BenchmarkSelectPk/01000-12                 54969             21824 ns/op            4648 B/op         97 allocs/op
BenchmarkSelectPk/01000-12                 54782             22163 ns/op            4648 B/op         97 allocs/op
BenchmarkSelectPk/10000-12                 54614             22786 ns/op            4648 B/op         97 allocs/op
BenchmarkSelectPk/10000-12                 56102             21710 ns/op            4648 B/op         97 allocs/op
BenchmarkSelectPk/10000-12                 54963             21659 ns/op            4648 B/op         97 allocs/op
BenchmarkSelectPk/10000-12                 55752             21549 ns/op            4648 B/op         97 allocs/op
BenchmarkSelectPk/10000-12                 55063             21579 ns/op            4648 B/op         97 allocs/op
```

Benchstat:

```bash
$ benchstat old.txt new.txt
name               old time/op    new time/op    delta
SelectPk/00001-12    17.0µs ± 3%    17.6µs ± 5%     ~     (p=0.056 n=5+5)
SelectPk/00010-12    19.0µs ± 2%    17.4µs ± 2%   -8.49%  (p=0.008 n=5+5)
SelectPk/00100-12    97.1µs ± 2%    22.2µs ± 2%  -77.10%  (p=0.008 n=5+5)
SelectPk/01000-12     888µs ± 2%      22µs ± 1%  -97.52%  (p=0.008 n=5+5)
SelectPk/10000-12    9.18ms ± 3%    0.02ms ± 4%  -99.76%  (p=0.008 n=5+5)

name               old alloc/op   new alloc/op   delta
SelectPk/00001-12    4.18kB ± 0%    4.63kB ± 0%  +10.73%  (p=0.016 n=4+5)
SelectPk/00010-12    5.22kB ± 0%    4.63kB ± 0%  -11.34%  (p=0.008 n=5+5)
SelectPk/00100-12    15.7kB ± 0%     4.6kB ± 0%  -70.47%  (p=0.008 n=5+5)
SelectPk/01000-12     138kB ± 0%       5kB ± 0%  -96.64%  (p=0.008 n=5+5)
SelectPk/10000-12    1.37MB ± 0%    0.00MB ± 0%  -99.66%  (p=0.008 n=5+5)

name               old allocs/op  new allocs/op  delta
SelectPk/00001-12      90.0 ± 0%      95.0 ± 0%   +5.56%  (p=0.008 n=5+5)
SelectPk/00010-12       135 ± 0%        95 ± 0%  -29.63%  (p=0.008 n=5+5)
SelectPk/00100-12       585 ± 0%        95 ± 0%  -83.76%  (p=0.008 n=5+5)
SelectPk/01000-12     6.83k ± 0%     0.10k ± 0%  -98.58%  (p=0.008 n=5+5)
SelectPk/10000-12     69.8k ± 0%      0.1k ± 0%  -99.86%  (p=0.008 n=5+5)
```


Patch:
diff --git a/database/index.go b/database/index.go
index c3b18a5f..6c474157 100644
--- a/database/index.go
+++ b/database/index.go
@@ -200,7 +200,7 @@ func (idx *Index) Truncate() error {
 // EncodeValue encodes the value we are going to use as a key,
 // If the index is typed, encode the value without expecting
 // the presence of other types.
-// Ff not, encode so that order is preserved regardless of the type.
+// If not, encode so that order is preserved regardless of the type.
 func (idx *Index) EncodeValue(v document.Value) ([]byte, error) {
 	if idx.Info.Type != 0 {
 		return v.MarshalBinary()
diff --git a/planner/explain.go b/planner/explain.go
index 95450803..7b363099 100644
--- a/planner/explain.go
+++ b/planner/explain.go
@@ -23,7 +23,7 @@ type ExplainStmt struct {
 func (s *ExplainStmt) Run(tx *database.Transaction, params []expr.Param) (query.Result, error) {
 	switch t := s.Statement.(type) {
 	case *Statement:
-		s, err := Optimize(t.Stream, tx)
+		s, err := Optimize(t.Stream, tx, params)
 		if err != nil {
 			return query.Result{}, err
 		}
diff --git a/planner/optimizer.go b/planner/optimizer.go
index 18010c1e..67b3dade 100644
--- a/planner/optimizer.go
+++ b/planner/optimizer.go
@@ -9,7 +9,7 @@ import (
 	"github.com/genjidb/genji/stream"
 )
 
-var optimizerRules = []func(s *stream.Stream, tx *database.Transaction) (*stream.Stream, error){
+var optimizerRules = []func(s *stream.Stream, tx *database.Transaction, params []expr.Param) (*stream.Stream, error){
 	SplitANDConditionRule,
 	PrecalculateExprRule,
 	RemoveUnnecessaryFilterNodesRule,
@@ -22,11 +22,11 @@ var optimizerRules = []func(s *stream.Stream, tx *database.Transaction) (*stream
 // and returns an optimized tree.
 // Depending on the rule, the tree may be modified in place or
 // replaced by a new one.
-func Optimize(s *stream.Stream, tx *database.Transaction) (*stream.Stream, error) {
+func Optimize(s *stream.Stream, tx *database.Transaction, params []expr.Param) (*stream.Stream, error) {
 	var err error
 
 	for _, rule := range optimizerRules {
-		s, err = rule(s, tx)
+		s, err = rule(s, tx, params)
 		if err != nil {
 			return nil, err
 		}
@@ -49,7 +49,7 @@ func Optimize(s *stream.Stream, tx *database.Transaction) (*stream.Stream, error
 //     filter(a > 2)
 //     filter(b != 3)
 //     filter(c < 2)
-func SplitANDConditionRule(s *stream.Stream, _ *database.Transaction) (*stream.Stream, error) {
+func SplitANDConditionRule(s *stream.Stream, _ *database.Transaction, _ []expr.Param) (*stream.Stream, error) {
 	n := s.Op
 
 	for n != nil {
@@ -103,16 +103,23 @@ func splitANDExpr(cond expr.Expr) (exprs []expr.Expr) {
 // Examples:
 //   3 + 4 --> 7
 //   3 + 1 > 10 - a --> 4 > 10 - a
-func PrecalculateExprRule(s *stream.Stream, _ *database.Transaction) (*stream.Stream, error) {
+func PrecalculateExprRule(s *stream.Stream, _ *database.Transaction, params []expr.Param) (*stream.Stream, error) {
 	n := s.Op
 
+	var err error
 	for n != nil {
 		switch t := n.(type) {
 		case *stream.FilterOperator:
-			t.E = precalculateExpr(t.E)
+			t.E, err = precalculateExpr(t.E, params)
+			if err != nil {
+				return nil, err
+			}
 		case *stream.ProjectOperator:
 			for i, e := range t.Exprs {
-				t.Exprs[i] = precalculateExpr(e)
+				t.Exprs[i], err = precalculateExpr(e, params)
+				if err != nil {
+					return nil, err
+				}
 			}
 		}
 
@@ -126,14 +133,17 @@ func PrecalculateExprRule(s *stream.Stream, _ *database.Transaction) (*stream.St
 // expression nodes when possible.
 // it returns a new expression with simplified nodes.
 // if no simplification is possible it returns the same expression.
-func precalculateExpr(e expr.Expr) expr.Expr {
+func precalculateExpr(e expr.Expr, params []expr.Param) (expr.Expr, error) {
 	switch t := e.(type) {
 	case expr.LiteralExprList:
 		// we assume that the list of expressions contains only literals
 		// until proven wrong.
 		literalsOnly := true
 		for i, te := range t {
-			newExpr := precalculateExpr(te)
+			newExpr, err := precalculateExpr(te, params)
+			if err != nil {
+				return nil, err
+			}
 			if _, ok := newExpr.(expr.LiteralValue); !ok {
 				literalsOnly = false
 			}
@@ -148,7 +158,7 @@ func precalculateExpr(e expr.Expr) expr.Expr {
 				values[i] = document.Value(t[i].(expr.LiteralValue))
 			}
 
-			return expr.ArrayValue(document.NewValueBuffer(values...))
+			return expr.ArrayValue(document.NewValueBuffer(values...)), nil
 		}
 
 	case *expr.KVPairs:
@@ -156,8 +166,12 @@ func precalculateExpr(e expr.Expr) expr.Expr {
 		// until proven wrong.
 		literalsOnly := true
 
+		var err error
 		for i, kv := range t.Pairs {
-			kv.V = precalculateExpr(kv.V)
+			kv.V, err = precalculateExpr(kv.V, params)
+			if err != nil {
+				return nil, err
+			}
 			if _, ok := kv.V.(expr.LiteralValue); !ok {
 				literalsOnly = false
 			}
@@ -173,7 +187,7 @@ func precalculateExpr(e expr.Expr) expr.Expr {
 				fb.Add(t.Pairs[i].K, document.Value(t.Pairs[i].V.(expr.LiteralValue)))
 			}
 
-			return expr.LiteralValue(document.NewDocumentValue(&fb))
+			return expr.LiteralValue(document.NewDocumentValue(&fb)), nil
 		}
 	case expr.Operator:
 		// since expr.Operator is an interface,
@@ -183,11 +197,17 @@ func precalculateExpr(e expr.Expr) expr.Expr {
 			!expr.IsOrOperator(t) &&
 			!expr.IsArithmeticOperator(t) &&
 			!expr.IsComparisonOperator(t) {
-			return e
+			return e, nil
 		}
 
-		lh := precalculateExpr(t.LeftHand())
-		rh := precalculateExpr(t.RightHand())
+		lh, err := precalculateExpr(t.LeftHand(), params)
+		if err != nil {
+			return nil, err
+		}
+		rh, err := precalculateExpr(t.RightHand(), params)
+		if err != nil {
+			return nil, err
+		}
 		t.SetLeftHandExpr(lh)
 		t.SetRightHandExpr(rh)
 
@@ -201,18 +221,24 @@ func precalculateExpr(e expr.Expr) expr.Expr {
 				panic(err)
 			}
 			// we replace this expression with the result of its evaluation
-			return expr.LiteralValue(v)
+			return expr.LiteralValue(v), nil
 		}
+	case expr.PositionalParam, expr.NamedParam:
+		v, err := e.Eval(&expr.Environment{Params: params})
+		if err != nil {
+			return nil, err
+		}
+		return expr.LiteralValue(v), nil
 	}
 
-	return e
+	return e, nil
 }
 
 // RemoveUnnecessaryFilterNodesRule removes any filter node whose
 // condition is a constant expression that evaluates to a truthy value.
 // if it evaluates to a falsy value, it considers that the tree
 // will not stream any document, so it returns an empty tree.
-func RemoveUnnecessaryFilterNodesRule(s *stream.Stream, _ *database.Transaction) (*stream.Stream, error) {
+func RemoveUnnecessaryFilterNodesRule(s *stream.Stream, _ *database.Transaction, _ []expr.Param) (*stream.Stream, error) {
 	n := s.Op
 
 	for n != nil {
@@ -263,7 +289,7 @@ func RemoveUnnecessaryFilterNodesRule(s *stream.Stream, _ *database.Transaction)
 
 // RemoveUnnecessaryProjection removes any project node whose
 // expression is a wildcard only.
-func RemoveUnnecessaryProjection(s *stream.Stream, _ *database.Transaction) (*stream.Stream, error) {
+func RemoveUnnecessaryProjection(s *stream.Stream, _ *database.Transaction, _ []expr.Param) (*stream.Stream, error) {
 	n := s.Op
 
 	for n != nil {
@@ -285,7 +311,7 @@ func RemoveUnnecessaryProjection(s *stream.Stream, _ *database.Transaction) (*st
 
 // RemoveUnnecessaryDistinctNodeRule removes any Dedup nodes
 // where projection is already unique.
-func RemoveUnnecessaryDistinctNodeRule(s *stream.Stream, tx *database.Transaction) (*stream.Stream, error) {
+func RemoveUnnecessaryDistinctNodeRule(s *stream.Stream, tx *database.Transaction, _ []expr.Param) (*stream.Stream, error) {
 	n := s.Op
 
 	// we assume that if we are reading from a table, the first
@@ -363,8 +389,8 @@ func isProjectionUnique(indexes database.Indexes, po *stream.ProjectOperator, pk
 // - one of its operands is a path expression that is indexed
 // - the other operand is a literal value or a parameter
 // If found, it will replace the input node by an indexInputNode using this index.
-// TODO(asdine): add support for ORDER BY and primary keys
-func UseIndexBasedOnFilterNodeRule(s *stream.Stream, tx *database.Transaction) (*stream.Stream, error) {
+// TODO(asdine): add support for ORDER BY
+func UseIndexBasedOnFilterNodeRule(s *stream.Stream, tx *database.Transaction, params []expr.Param) (*stream.Stream, error) {
 	n := s.Op
 
 	// first we lookup for the seq scan node.
@@ -384,29 +410,20 @@ func UseIndexBasedOnFilterNodeRule(s *stream.Stream, tx *database.Transaction) (
 		return nil, err
 	}
 
+	info := t.Info()
 	indexes := t.Indexes()
 
-	type candidate struct {
-		filterOp *stream.FilterOperator
-		in       *stream.IndexScanOperator
-		index    *database.Index
-	}
-
-	var candidates []candidate
+	var candidates []*candidate
 
 	// look for all selection nodes that satisfy our requirements
 	for n != nil {
 		if f, ok := n.(*stream.FilterOperator); ok {
-			indexedNode, idx, err := filterNodeValidForIndex(f, st.TableName, indexes)
+			candidate, err := getCandidateFromfilterNode(f, st.TableName, info, indexes)
 			if err != nil {
 				return nil, err
 			}
-			if indexedNode != nil {
-				candidates = append(candidates, candidate{
-					filterOp: f,
-					in:       indexedNode,
-					index:    idx,
-				})
+			if candidate != nil {
+				candidates = append(candidates, candidate)
 			}
 		}
 
@@ -420,23 +437,23 @@ func UseIndexBasedOnFilterNodeRule(s *stream.Stream, tx *database.Transaction) (
 	var cost int
 
 	for i, candidate := range candidates {
-		currentCost := candidate.in.Ranges.Cost()
+		currentCost := candidate.cost
 
 		if selectedCandidate == nil {
-			selectedCandidate = &candidates[i]
+			selectedCandidate = candidates[i]
 			cost = currentCost
 			continue
 		}
 
 		if currentCost < cost {
-			selectedCandidate = &candidates[i]
+			selectedCandidate = candidates[i]
 			cost = currentCost
 		}
 
-		// if the cost is the same and the candidate's related index is a unique index,
+		// if the cost is the same and the candidate's related index has a higher priority,
 		// select it.
-		if currentCost == cost && candidate.index.Info.Unique {
-			selectedCandidate = &candidates[i]
+		if currentCost == cost && selectedCandidate.priority < candidate.priority {
+			selectedCandidate = candidates[i]
 		}
 	}
 
@@ -448,96 +465,114 @@ func UseIndexBasedOnFilterNodeRule(s *stream.Stream, tx *database.Transaction) (
 	s.Remove(selectedCandidate.filterOp)
 
 	// we replace the seq scan node by the selected index scan node
-	stream.InsertBefore(s.First(), selectedCandidate.in)
+	stream.InsertBefore(s.First(), selectedCandidate.newOp)
 
 	s.Remove(s.First().GetNext())
 
 	return s, nil
 }
 
-func filterNodeValidForIndex(sn *stream.FilterOperator, tableName string, indexes database.Indexes) (*stream.IndexScanOperator, *database.Index, error) {
-	if sn.E == nil {
-		return nil, nil, nil
+type candidate struct {
+	// filter operator to remove and replace by either an indexScan
+	// or pkScan operators.
+	filterOp *stream.FilterOperator
+	// the candidate indexScan or pkScan operator
+	newOp stream.Operator
+	// the cost of the candidate
+	cost int
+	// is this candidate reading from an index
+	isIndex bool
+	// is this candidate reading primary key ranges
+	isPk bool
+	// if the costs of two candidates are equal,
+	// this number determines which node will be prioritized
+	priority int
+}
+
+// getCandidateFromfilterNode analyses f and determines if it can be replaced by an indexScan or pkScan operator.
+func getCandidateFromfilterNode(f *stream.FilterOperator, tableName string, info *database.TableInfo, indexes database.Indexes) (*candidate, error) {
+	if f.E == nil {
+		return nil, nil
 	}
 
 	// the root of the condition must be an operator
-	op, ok := sn.E.(expr.Operator)
+	op, ok := f.E.(expr.Operator)
 	if !ok {
-		return nil, nil, nil
+		return nil, nil
 	}
 
 	// determine if the operator can read from the index
 	if !expr.OperatorIsIndexCompatible(op) {
-		return nil, nil, nil
+		return nil, nil
 	}
 
 	// determine if the operator can benefit from an index
 	ok, path, e := opCanUseIndex(op)
 	if !ok {
-		return nil, nil, nil
+		return nil, nil
 	}
 
-	// analyse the other operand to make sure it's a literal or a param
-	if !isLiteralOrParam(e) {
-		return nil, nil, nil
+	// analyse the other operand to make sure it's a literal
+	ev, ok := e.(expr.LiteralValue)
+	if !ok {
+		return nil, nil
 	}
+	v := document.Value(ev)
 
 	// now, we look if an index exists for that path
-	idx := indexes.GetIndexByPath(document.Path(path))
-	if idx == nil {
-		return nil, nil, nil
+	cd := candidate{
+		filterOp: f,
 	}
 
-	var ranges []stream.Range
+	// we'll start with checking if the path is the primary key of the table
+	if pk := info.GetPrimaryKey(); pk != nil && pk.Path.IsEqual(path) {
+		// if both types are different, don't select this scanner
+		if pk.Type != v.Type {
+			return nil, nil
+		}
 
-	switch op.(type) {
-	case *expr.EqOperator:
-		ranges = append(ranges, stream.Range{
-			Min:   e,
-			Exact: true,
-		})
-	case *expr.GtOperator:
-		ranges = append(ranges, stream.Range{
-			Min:       e,
-			Exclusive: true,
-		})
-	case *expr.GteOperator:
-		ranges = append(ranges, stream.Range{
-			Min: e,
-		})
-	case *expr.LtOperator:
-		ranges = append(ranges, stream.Range{
-			Max:       e,
-			Exclusive: true,
-		})
-	case *expr.LteOperator:
-		ranges = append(ranges, stream.Range{
-			Max: e,
-		})
-	case *expr.InOperator:
-		// opCanUseIndex made sure e is an array.
-		a := e.(expr.LiteralValue).V.(document.Array)
-		err := a.Iterate(func(i int, value document.Value) error {
-			ranges = append(ranges, stream.Range{
-				Min:   expr.LiteralValue(value),
-				Exact: true,
-			})
-			return nil
-		})
+		cd.isPk = true
+		cd.priority = 3
+
+		ranges, err := getRangesFromOp(op, v)
 		if err != nil {
-			return nil, nil, err
+			return nil, err
 		}
-	default:
-		panic(fmt.Sprintf("unknown operator %#v", op))
+
+		cd.newOp = stream.PkScan(tableName, ranges...)
+		cd.cost = ranges.Cost()
+		return &cd, nil
 	}
 
-	node := stream.IndexScan(idx.Info.IndexName)
-	node.Ranges = ranges
+	// if not, check if an index exists for that path
+	if idx := indexes.GetIndexByPath(document.Path(path)); idx != nil {
+		// if both types are different, don't select this scanner
+		if !idx.Info.Type.IsZero() && idx.Info.Type != v.Type {
+			return nil, nil
+		}
+
+		cd.isIndex = true
+		if idx.Info.Unique {
+			cd.priority = 2
+		} else {
+			cd.priority = 1
+		}
 
-	return node, idx, nil
+		ranges, err := getRangesFromOp(op, v)
+		if err != nil {
+			return nil, err
+		}
+
+		cd.newOp = stream.IndexScan(idx.Info.IndexName, ranges...)
+		cd.cost = ranges.Cost()
+
+		return &cd, nil
+	}
+
+	return nil, nil
 }
 
-func opCanUseIndex(op expr.Operator) (bool, expr.Path, expr.Expr) {
+func opCanUseIndex(op expr.Operator) (bool, document.Path, expr.Expr) {
 	lf, leftIsField := op.LeftHand().(expr.Path)
 	rf, rightIsField := op.RightHand().(expr.Path)
 
@@ -555,7 +590,7 @@ func opCanUseIndex(op expr.Operator) (bool, expr.Path, expr.Expr) {
 				return false, nil, nil
 			}
 
-			return true, lf, rh
+			return true, document.Path(lf), rh
 		}
 
 		return false, nil, nil
@@ -563,22 +598,60 @@ func opCanUseIndex(op expr.Operator) (bool, expr.Path, expr.Expr) {
 
 	// path OP expr
 	if leftIsField && !rightIsField {
-		return true, lf, op.RightHand()
+		return true, document.Path(lf), op.RightHand()
 	}
 
 	// expr OP path
 	if rightIsField && !leftIsField {
-		return true, rf, op.LeftHand()
+		return true, document.Path(rf), op.LeftHand()
 	}
 
 	return false, nil, nil
 }
 
-func isLiteralOrParam(e expr.Expr) (ok bool) {
-	switch e.(type) {
-	case expr.LiteralValue, expr.NamedParam, expr.PositionalParam:
-		return true
+func getRangesFromOp(op expr.Operator, v document.Value) (stream.Ranges, error) {
+	var ranges stream.Ranges
+
+	switch op.(type) {
+	case *expr.EqOperator:
+		ranges = ranges.Append(stream.Range{
+			Min:   v,
+			Exact: true,
+		})
+	case *expr.GtOperator:
+		ranges = ranges.Append(stream.Range{
+			Min:       v,
+			Exclusive: true,
+		})
+	case *expr.GteOperator:
+		ranges = ranges.Append(stream.Range{
+			Min: v,
+		})
+	case *expr.LtOperator:
+		ranges = ranges.Append(stream.Range{
+			Max:       v,
+			Exclusive: true,
+		})
+	case *expr.LteOperator:
+		ranges = ranges.Append(stream.Range{
+			Max: v,
+		})
+	case *expr.InOperator:
+		// opCanUseIndex made sure e is an array.
+		a := v.V.(document.Array)
+		err := a.Iterate(func(i int, value document.Value) error {
+			ranges = ranges.Append(stream.Range{
+				Min:   value,
+				Exact: true,
+			})
+			return nil
+		})
+		if err != nil {
+			return nil, err
+		}
+	default:
+		panic(fmt.Sprintf("unknown operator %#v", op))
 	}
 
-	return false
+	return ranges, nil
 }
diff --git a/planner/statement.go b/planner/statement.go
index 986eabce..2cd3ed9f 100644
--- a/planner/statement.go
+++ b/planner/statement.go
@@ -17,7 +17,7 @@ type Statement struct {
 // Run returns a result containing the stream. The stream will be executed by calling the Iterate method of
 // the result.
 func (s *Statement) Run(tx *database.Transaction, params []expr.Param) (query.Result, error) {
-	st, err := Optimize(s.Stream, tx)
+	st, err := Optimize(s.Stream, tx, params)
 	if err != nil || st == nil {
 		return query.Result{}, err
 	}
diff --git a/stream/iterator.go b/stream/iterator.go
index e28b9f2c..8104262a 100644
--- a/stream/iterator.go
+++ b/stream/iterator.go
@@ -209,7 +209,7 @@ func (it *PkScanOperator) Iterate(in *expr.Environment, fn func(out *expr.Enviro
 		return err
 	}
 
-	ranges, err := it.Ranges.Encode(table, in)
+	err = it.Ranges.Encode(table, in)
 	if err != nil {
 		return err
 	}
@@ -222,14 +222,14 @@ func (it *PkScanOperator) Iterate(in *expr.Environment, fn func(out *expr.Enviro
 		iterator = table.DescendLessOrEqual
 	}
 
-	for _, rng := range ranges {
+	for _, rng := range it.Ranges {
 		var start, end document.Value
 		if !it.Reverse {
-			start = rng.minVal
-			end = rng.maxVal
+			start = rng.Min
+			end = rng.Max
 		} else {
-			start = rng.maxVal
-			end = rng.minVal
+			start = rng.Max
+			end = rng.Min
 		}
 
 		var encEnd []byte
@@ -303,12 +303,7 @@ func (it *IndexScanOperator) String() string {
 	s.WriteString(strconv.Quote(it.IndexName))
 	if len(it.Ranges) > 0 {
 		s.WriteString(", ")
-		for i, r := range it.Ranges {
-			s.WriteString(r.String())
-			if i+1 < len(it.Ranges) {
-				s.WriteString(", ")
-			}
-		}
+		s.WriteString(it.Ranges.String())
 	}
 
 	s.WriteString(")")
@@ -332,7 +327,7 @@ func (it *IndexScanOperator) Iterate(in *expr.Environment, fn func(out *expr.Env
 		return err
 	}
 
-	ranges, err := it.Ranges.Encode(index, in)
+	err = it.Ranges.Encode(index, in)
 	if err != nil {
 		return err
 	}
@@ -358,14 +353,14 @@ func (it *IndexScanOperator) Iterate(in *expr.Environment, fn func(out *expr.Env
 		})
 	}
 
-	for _, rng := range ranges {
+	for _, rng := range it.Ranges {
 		var start, end document.Value
 		if !it.Reverse {
-			start = rng.minVal
-			end = rng.maxVal
+			start = rng.Min
+			end = rng.Max
 		} else {
-			start = rng.maxVal
-			end = rng.minVal
+			start = rng.Max
+			end = rng.Min
 		}
 
 		var encEnd []byte
@@ -412,7 +407,7 @@ func (it *IndexScanOperator) Iterate(in *expr.Environment, fn func(out *expr.Env
 }
 
 type Range struct {
-	Min, Max expr.Expr
+	Min, Max document.Value
 	// Exclude Min and Max from the results.
 	// By default, min and max are inclusive.
 	// Exclusive and Exact cannot be set to true at the same time.
@@ -421,51 +416,43 @@ type Range struct {
 	// If set to true, Max will be ignored for comparison
 	// and for determining the global upper bound.
 	Exact bool
+
+	encodedMin, encodedMax []byte
+	rangeType              document.ValueType
 }
 
-func (r *Range) encode(er *encodedRange, encoder ValueEncoder, env *expr.Environment) error {
+func (r *Range) encode(encoder ValueEncoder, env *expr.Environment) error {
+	var err error
+
 	// first we evaluate Min and Max
-	if r.Min != nil {
-		minVal, err := r.Min.Eval(env)
-		if err != nil {
-			return err
-		}
-		er.minVal = minVal
-		er.min, err = encoder.EncodeValue(minVal)
+	if !r.Min.Type.IsZero() {
+		r.encodedMin, err = encoder.EncodeValue(r.Min)
 		if err != nil {
 			return err
 		}
-		er.rangeType = er.minVal.Type
+		r.rangeType = r.Min.Type
 	}
-	if r.Max != nil {
-		maxVal, err := r.Max.Eval(env)
+	if !r.Max.Type.IsZero() {
+		r.encodedMax, err = encoder.EncodeValue(r.Max)
 		if err != nil {
 			return err
 		}
-		er.maxVal = maxVal
-		er.max, err = encoder.EncodeValue(maxVal)
-		if err != nil {
-			return err
-		}
-		if !er.rangeType.IsZero() && er.rangeType != maxVal.Type {
-			panic("range contain values of different type")
+		if !r.rangeType.IsZero() && r.rangeType != r.Max.Type {
+			panic("range contain values of different types")
 		}
 
-		er.rangeType = er.maxVal.Type
+		r.rangeType = r.Max.Type
 	}
 
 	// ensure boundaries are typed
-	if er.minVal.Type.IsZero() {
-		er.minVal.Type = er.rangeType
+	if r.Min.Type.IsZero() {
+		r.Min.Type = r.rangeType
 	}
-	if er.maxVal.Type.IsZero() {
-		er.maxVal.Type = er.rangeType
+	if r.Max.Type.IsZero() {
+		r.Max.Type = r.rangeType
 	}
 
-	er.exclusive = r.Exclusive
-	er.exact = r.Exact
-
-	if er.exclusive && er.exact {
+	if r.Exclusive && r.Exact {
 		panic("exclusive and exact cannot both be true")
 	}
 
@@ -477,37 +464,87 @@ func (r *Range) String() string {
 		return fmt.Sprintf("%v", r.Min)
 	}
 
-	min, max := r.Min, r.Max
-	if min == nil {
-		min = expr.IntegerValue(-1)
+	if r.Min.Type.IsZero() {
+		r.Min = document.NewIntegerValue(-1)
 	}
-	if max == nil {
-		max = expr.IntegerValue(-1)
+	if r.Max.Type.IsZero() {
+		r.Max = document.NewIntegerValue(-1)
 	}
 
 	if r.Exclusive {
-		return fmt.Sprintf("[%v, %v, true]", min, max)
+		return fmt.Sprintf("[%v, %v, true]", r.Min, r.Max)
 	}
 
-	return fmt.Sprintf("[%v, %v]", min, max)
+	return fmt.Sprintf("[%v, %v]", r.Min, r.Max)
+}
+
+func (r *Range) IsEqual(other *Range) bool {
+	if r.Exact != other.Exact {
+		return false
+	}
+
+	if r.rangeType != r.rangeType {
+		return false
+	}
+
+	if r.Exclusive != r.Exclusive {
+		return false
+	}
+
+	if r.Min.Type != other.Min.Type {
+		return false
+	}
+	ok, err := r.Min.IsEqual(other.Min)
+	if err != nil || !ok {
+		return false
+	}
+
+	if r.Max.Type != other.Max.Type {
+		return false
+	}
+	ok, err = r.Max.IsEqual(other.Max)
+	if err != nil || !ok {
+		return false
+	}
+
+	return true
 }
 
 type Ranges []Range
 
+// Append rng to r and return the new slice.
+// Duplicate ranges are ignored.
+func (r Ranges) Append(rng Range) Ranges {
+	// ensure we don't keep duplicate ranges
+	isDuplicate := false
+	for _, e := range r {
+		if e.IsEqual(&rng) {
+			isDuplicate = true
+			break
+		}
+	}
+
+	if isDuplicate {
+		return r
+	}
+
+	return append(r, rng)
+}
+
 type ValueEncoder interface {
 	EncodeValue(v document.Value) ([]byte, error)
 }
 
-func (r Ranges) Encode(encoder ValueEncoder, env *expr.Environment) (encodedRanges, error) {
-	enc := make([]encodedRange, len(r))
+// Encode each range using the given value encoder.
+func (r Ranges) Encode(encoder ValueEncoder, env *expr.Environment) error {
 	for i := range r {
-		err := r[i].encode(&enc[i], encoder, env)
+		err := r[i].encode(encoder, env)
 		if err != nil {
-			return nil, err
+			return err
 		}
 	}
 
-	return enc, nil
+	return nil
 }
 
 func (r Ranges) String() string {
@@ -538,12 +575,12 @@ func (r Ranges) Cost() int {
 		}
 
 		// if there are two boundaries, increment by 50
-		if rng.Min != nil && rng.Max != nil {
+		if !rng.Min.Type.IsZero() && !rng.Max.Type.IsZero() {
 			cost += 50
 		}
 
 		// if there is only one boundary, increment by 100
-		if (rng.Min != nil && rng.Max == nil) || (rng.Min == nil && rng.Max != nil) {
+		if (!rng.Min.Type.IsZero() && rng.Max.Type.IsZero()) || (rng.Min.Type.IsZero() && !rng.Max.Type.IsZero()) {
 			cost += 100
 			continue
 		}
@@ -555,46 +592,36 @@ func (r Ranges) Cost() int {
 	return cost
 }
 
-type encodedRange struct {
-	minVal, maxVal document.Value
-	rangeType      document.ValueType
-	min, max       []byte
-	exclusive      bool
-	exact          bool
-}
-
-func (e *encodedRange) IsInRange(value []byte) bool {
+func (r *Range) IsInRange(value []byte) bool {
 	// by default, we consider the value within range
 	cmpMin, cmpMax := 1, -1
 
 	// we compare with the lower bound and see if it matches
-	if e.min != nil {
-		cmpMin = bytes.Compare(value, e.min)
+	if r.encodedMin != nil {
+		cmpMin = bytes.Compare(value, r.encodedMin)
 	}
 
 	// if exact is true the value has to be equal to the lower bound.
-	if e.exact {
+	if r.Exact {
 		return cmpMin == 0
 	}
 
 	// if exclusive and the value is equal to the lower bound
 	// we can ignore it
-	if e.exclusive && cmpMin == 0 {
+	if r.Exclusive && cmpMin == 0 {
 		return false
 	}
 
 	// the value is bigger than the lower bound,
 	// see if it matches the upper bound.
-	if e.max != nil {
-		cmpMax = bytes.Compare(value, e.max)
+	if r.encodedMax != nil {
+		cmpMax = bytes.Compare(value, r.encodedMax)
 	}
 
 	// if boundaries are strict, ignore values equal to the max
-	if e.exclusive && cmpMax == 0 {
+	if r.Exclusive && cmpMax == 0 {
 		return false
 	}
 
 	return cmpMax <= 0
 }
-
-type encodedRanges []encodedRange


Test patch:
diff --git a/db_test.go b/db_test.go
index 482a1caa..219118ab 100644
--- a/db_test.go
+++ b/db_test.go
@@ -163,3 +163,26 @@ func BenchmarkSelectWhere(b *testing.B) {
 		})
 	}
 }
+
+func BenchmarkSelectPk(b *testing.B) {
+	for size := 1; size <= 10000; size *= 10 {
+		b.Run(fmt.Sprintf("%.05d", size), func(b *testing.B) {
+			db, err := genji.Open(":memory:")
+			require.NoError(b, err)
+
+			err = db.Exec("CREATE TABLE foo(a INT PRIMARY KEY)")
+			require.NoError(b, err)
+
+			for i := 0; i < size; i++ {
+				err = db.Exec("INSERT INTO foo(a) VALUES (?)", i)
+				require.NoError(b, err)
+			}
+
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				res, _ := db.Query("SELECT * FROM foo WHERE a = ?", size-1)
+				res.Iterate(func(d document.Document) error { return nil })
+			}
+		})
+	}
+}
diff --git a/planner/optimizer_test.go b/planner/optimizer_test.go
index 215ee7f7..abb06d4c 100644
--- a/planner/optimizer_test.go
+++ b/planner/optimizer_test.go
@@ -83,7 +83,7 @@ func TestSplitANDConditionRule(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			res, err := planner.SplitANDConditionRule(test.in, nil)
+			res, err := planner.SplitANDConditionRule(test.in, nil, nil)
 			require.NoError(t, err)
 			require.Equal(t, res.String(), test.expected.String())
 		})
@@ -94,26 +94,31 @@ func TestPrecalculateExprRule(t *testing.T) {
 	tests := []struct {
 		name        string
 		e, expected expr.Expr
+		params      []expr.Param
 	}{
 		{
 			"constant expr: 3 -> 3",
 			expr.IntegerValue(3),
 			expr.IntegerValue(3),
+			nil,
 		},
 		{
 			"operator with two constant operands: 3 + 2.4 -> 5.4",
-			expr.Add(expr.IntegerValue(3), expr.DoubleValue(2.4)),
+			expr.Add(expr.IntegerValue(3), expr.PositionalParam(1)),
 			expr.DoubleValue(5.4),
+			[]expr.Param{{Value: 2.4}},
 		},
 		{
 			"operator with constant nested operands: 3 > 1 - 40 -> true",
 			expr.Gt(expr.DoubleValue(3), expr.Sub(expr.IntegerValue(1), expr.DoubleValue(40))),
 			expr.BoolValue(true),
+			nil,
 		},
 		{
 			"constant sub-expr: a > 1 - 40 -> a > -39",
 			expr.Gt(expr.Path{document.PathFragment{FieldName: "a"}}, expr.Sub(expr.IntegerValue(1), expr.DoubleValue(40))),
 			expr.Gt(expr.Path{document.PathFragment{FieldName: "a"}}, expr.DoubleValue(-39)),
+			nil,
 		},
 		{
 			"constant sub-expr: a IN [1, 2] -> a IN array([1, 2])",
@@ -121,6 +126,7 @@ func TestPrecalculateExprRule(t *testing.T) {
 			expr.In(expr.Path{document.PathFragment{FieldName: "a"}}, expr.LiteralValue(document.NewArrayValue(document.NewValueBuffer().
 				Append(document.NewIntegerValue(1)).
 				Append(document.NewIntegerValue(2))))),
+			nil,
 		},
 		{
 			"non-constant expr list: [a, 1 - 40] -> [a, -39]",
@@ -132,6 +138,7 @@ func TestPrecalculateExprRule(t *testing.T) {
 				expr.Path{document.PathFragment{FieldName: "a"}},
 				expr.DoubleValue(-39),
 			},
+			nil,
 		},
 		{
 			"constant expr list: [3, 1 - 40] -> array([3, -39])",
@@ -142,6 +149,7 @@ func TestPrecalculateExprRule(t *testing.T) {
 			expr.LiteralValue(document.NewArrayValue(document.NewValueBuffer().
 				Append(document.NewIntegerValue(3)).
 				Append(document.NewDoubleValue(-39)))),
+			nil,
 		},
 		{
 			`non-constant kvpair: {"a": d, "b": 1 - 40} -> {"a": 3, "b": -39}`,
@@ -153,6 +161,7 @@ func TestPrecalculateExprRule(t *testing.T) {
 				{K: "a", V: expr.Path{document.PathFragment{FieldName: "d"}}},
 				{K: "b", V: expr.DoubleValue(-39)},
 			}},
+			nil,
 		},
 		{
 			`constant kvpair: {"a": 3, "b": 1 - 40} -> document({"a": 3, "b": -39})`,
@@ -164,6 +173,7 @@ func TestPrecalculateExprRule(t *testing.T) {
 				Add("a", document.NewIntegerValue(3)).
 				Add("b", document.NewDoubleValue(-39)),
 			)),
+			nil,
 		},
 	}
 
@@ -171,7 +181,7 @@ func TestPrecalculateExprRule(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			s := stream.New(stream.SeqScan("foo")).
 				Pipe(stream.Filter(test.e))
-			res, err := planner.PrecalculateExprRule(s, nil)
+			res, err := planner.PrecalculateExprRule(s, nil, test.params)
 			require.NoError(t, err)
 			require.Equal(t, stream.New(stream.SeqScan("foo")).Pipe(stream.Filter(test.expected)).String(), res.String())
 		})
@@ -210,7 +220,7 @@ func TestRemoveUnnecessarySelectionNodesRule(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			res, err := planner.RemoveUnnecessaryFilterNodesRule(test.root, nil)
+			res, err := planner.RemoveUnnecessaryFilterNodesRule(test.root, nil, nil)
 			require.NoError(t, err)
 			if test.expected != nil {
 				require.Equal(t, test.expected.String(), res.String())
@@ -289,7 +299,7 @@ func TestRemoveUnnecessaryDedupNodeRule(t *testing.T) {
 			`)
 			require.NoError(t, err)
 
-			res, err := planner.RemoveUnnecessaryDistinctNodeRule(test.root, tx.Transaction)
+			res, err := planner.RemoveUnnecessaryDistinctNodeRule(test.root, tx.Transaction, nil)
 			require.NoError(t, err)
 			require.Equal(t, test.expected.String(), res.String())
 		})
@@ -309,14 +319,14 @@ func TestUseIndexBasedOnSelectionNodeRule(t *testing.T) {
 		{
 			"FROM foo WHERE a = 1",
 			stream.New(stream.SeqScan("foo")).Pipe(stream.Filter(parser.MustParseExpr("a = 1"))),
-			stream.New(stream.IndexScan("idx_foo_a", st.Range{Min: parser.MustParseExpr("1"), Exact: true})),
+			stream.New(stream.IndexScan("idx_foo_a", st.Range{Min: document.NewIntegerValue(1), Exact: true})),
 		},
 		{
 			"FROM foo WHERE a = 1 AND b = 2",
 			stream.New(stream.SeqScan("foo")).
 				Pipe(stream.Filter(parser.MustParseExpr("a = 1"))).
 				Pipe(stream.Filter(parser.MustParseExpr("b = 2"))),
-			stream.New(stream.IndexScan("idx_foo_b", st.Range{Min: parser.MustParseExpr("2"), Exact: true})).
+			stream.New(stream.IndexScan("idx_foo_b", st.Range{Min: document.NewIntegerValue(2), Exact: true})).
 				Pipe(stream.Filter(parser.MustParseExpr("a = 1"))),
 		},
 		{
@@ -324,7 +334,7 @@ func TestUseIndexBasedOnSelectionNodeRule(t *testing.T) {
 			stream.New(stream.SeqScan("foo")).
 				Pipe(stream.Filter(parser.MustParseExpr("c = 3"))).
 				Pipe(stream.Filter(parser.MustParseExpr("b = 2"))),
-			stream.New(stream.IndexScan("idx_foo_c", st.Range{Min: parser.MustParseExpr("3"), Exact: true})).
+			stream.New(stream.IndexScan("idx_foo_c", st.Range{Min: document.NewIntegerValue(3), Exact: true})).
 				Pipe(stream.Filter(parser.MustParseExpr("b = 2"))),
 		},
 		{
@@ -332,7 +342,7 @@ func TestUseIndexBasedOnSelectionNodeRule(t *testing.T) {
 			stream.New(stream.SeqScan("foo")).
 				Pipe(stream.Filter(parser.MustParseExpr("c > 3"))).
 				Pipe(stream.Filter(parser.MustParseExpr("b = 2"))),
-			stream.New(stream.IndexScan("idx_foo_b", st.Range{Min: parser.MustParseExpr("2"), Exact: true})).
+			stream.New(stream.IndexScan("idx_foo_b", st.Range{Min: document.NewIntegerValue(2), Exact: true})).
 				Pipe(stream.Filter(parser.MustParseExpr("c > 3"))),
 		},
 		{
@@ -341,10 +351,31 @@ func TestUseIndexBasedOnSelectionNodeRule(t *testing.T) {
 				Pipe(stream.Filter(parser.MustParseExpr("c = 3"))).
 				Pipe(stream.Filter(parser.MustParseExpr("b = 2"))).
 				Pipe(stream.Project(parser.MustParseExpr("a"))),
-			stream.New(stream.IndexScan("idx_foo_c", st.Range{Min: parser.MustParseExpr("3"), Exact: true})).
+			stream.New(stream.IndexScan("idx_foo_c", st.Range{Min: document.NewIntegerValue(3), Exact: true})).
 				Pipe(stream.Filter(parser.MustParseExpr("b = 2"))).
 				Pipe(stream.Project(parser.MustParseExpr("a"))),
 		},
+		{
+			"SELECT a FROM foo WHERE c = 'hello' AND b = 2",
+			stream.New(stream.SeqScan("foo")).
+				Pipe(stream.Filter(parser.MustParseExpr("c = 'hello'"))).
+				Pipe(stream.Filter(parser.MustParseExpr("b = 2"))).
+				Pipe(stream.Project(parser.MustParseExpr("a"))),
+			stream.New(stream.IndexScan("idx_foo_b", st.Range{Min: document.NewIntegerValue(2), Exact: true})).
+				Pipe(stream.Filter(parser.MustParseExpr("c = 'hello'"))).
+				Pipe(stream.Project(parser.MustParseExpr("a"))),
+		},
+		{
+			"SELECT a FROM foo WHERE c = 'hello' AND d = 2",
+			stream.New(stream.SeqScan("foo")).
+				Pipe(stream.Filter(parser.MustParseExpr("c = 'hello'"))).
+				Pipe(stream.Filter(parser.MustParseExpr("d = 2"))).
+				Pipe(stream.Project(parser.MustParseExpr("a"))),
+			stream.New(stream.SeqScan("foo")).
+				Pipe(stream.Filter(parser.MustParseExpr("c = 'hello'"))).
+				Pipe(stream.Filter(parser.MustParseExpr("d = 2"))).
+				Pipe(stream.Project(parser.MustParseExpr("a"))),
+		},
 		{
 			"FROM foo WHERE a IN [1, 2]",
 			stream.New(stream.SeqScan("foo")).Pipe(stream.Filter(
@@ -353,7 +384,7 @@ func TestUseIndexBasedOnSelectionNodeRule(t *testing.T) {
 					expr.ArrayValue(document.NewValueBuffer(document.NewIntegerValue(1), document.NewIntegerValue(2))),
 				),
 			)),
-			stream.New(stream.IndexScan("idx_foo_a", st.Range{Min: parser.MustParseExpr("1"), Exact: true}, st.Range{Min: parser.MustParseExpr("2"), Exact: true})),
+			stream.New(stream.IndexScan("idx_foo_a", st.Range{Min: document.NewIntegerValue(1), Exact: true}, st.Range{Min: document.NewIntegerValue(2), Exact: true})),
 		},
 		{
 			"FROM foo WHERE 1 IN a",
@@ -361,9 +392,46 @@ func TestUseIndexBasedOnSelectionNodeRule(t *testing.T) {
 			stream.New(stream.SeqScan("foo")).Pipe(stream.Filter(parser.MustParseExpr("1 IN a"))),
 		},
 		{
-			"FROM foo WHERE a > 10",
-			stream.New(stream.SeqScan("foo")).Pipe(stream.Filter(parser.MustParseExpr("a > 10"))),
-			stream.New(stream.IndexScan("idx_foo_a", st.Range{Min: parser.MustParseExpr("10"), Exclusive: true})),
+			"FROM foo WHERE a >= 10",
+			stream.New(stream.SeqScan("foo")).Pipe(stream.Filter(parser.MustParseExpr("a >= 10"))),
+			stream.New(stream.IndexScan("idx_foo_a", st.Range{Min: document.NewIntegerValue(10)})),
+		},
+		{
+			"FROM foo WHERE k = 1",
+			stream.New(stream.SeqScan("foo")).Pipe(stream.Filter(parser.MustParseExpr("k = 1"))),
+			stream.New(stream.PkScan("foo", st.Range{Min: document.NewIntegerValue(1), Exact: true})),
+		},
+		{
+			"FROM foo WHERE k = 1 AND b = 2",
+			stream.New(stream.SeqScan("foo")).
+				Pipe(stream.Filter(parser.MustParseExpr("k = 1"))).
+				Pipe(stream.Filter(parser.MustParseExpr("b = 2"))),
+			stream.New(stream.PkScan("foo", st.Range{Min: document.NewIntegerValue(1), Exact: true})).
+				Pipe(stream.Filter(parser.MustParseExpr("b = 2"))),
+		},
+		{
+			"FROM foo WHERE a = 1 AND k = 2",
+			stream.New(stream.SeqScan("foo")).
+				Pipe(stream.Filter(parser.MustParseExpr("a = 1"))).
+				Pipe(stream.Filter(parser.MustParseExpr("2 = k"))),
+			stream.New(stream.PkScan("foo", st.Range{Min: document.NewIntegerValue(2), Exact: true})).
+				Pipe(stream.Filter(parser.MustParseExpr("a = 1"))),
+		},
+		{
+			"FROM foo WHERE a = 1 AND k < 2",
+			stream.New(stream.SeqScan("foo")).
+				Pipe(stream.Filter(parser.MustParseExpr("a = 1"))).
+				Pipe(stream.Filter(parser.MustParseExpr("k < 2"))),
+			stream.New(stream.IndexScan("idx_foo_a", st.Range{Min: document.NewIntegerValue(1), Exact: true})).
+				Pipe(stream.Filter(parser.MustParseExpr("k < 2"))),
+		},
+		{
+			"FROM foo WHERE a = 1 AND k = 'hello'",
+			stream.New(stream.SeqScan("foo")).
+				Pipe(stream.Filter(parser.MustParseExpr("a = 1"))).
+				Pipe(stream.Filter(parser.MustParseExpr("k = 'hello'"))),
+			stream.New(stream.IndexScan("idx_foo_a", st.Range{Min: document.NewIntegerValue(1), Exact: true})).
+				Pipe(stream.Filter(parser.MustParseExpr("k = 'hello'"))),
 		},
 	}
 
@@ -378,7 +446,7 @@ func TestUseIndexBasedOnSelectionNodeRule(t *testing.T) {
 			defer tx.Rollback()
 
 			err = tx.Exec(`
-				CREATE TABLE foo (k INT PRIMARY KEY);
+				CREATE TABLE foo (k INT PRIMARY KEY, c INT);
 				CREATE INDEX idx_foo_a ON foo(a);
 				CREATE INDEX idx_foo_b ON foo(b);
 				CREATE UNIQUE INDEX idx_foo_c ON foo(c);
@@ -389,7 +457,7 @@ func TestUseIndexBasedOnSelectionNodeRule(t *testing.T) {
 			`)
 			require.NoError(t, err)
 
-			res, err := planner.UseIndexBasedOnFilterNodeRule(test.root, tx.Transaction)
+			res, err := planner.UseIndexBasedOnFilterNodeRule(test.root, tx.Transaction, nil)
 			require.NoError(t, err)
 			require.Equal(t, test.expected.String(), res.String())
 		})
diff --git a/stream/iterator_test.go b/stream/iterator_test.go
index 2711124c..8162fddc 100644
--- a/stream/iterator_test.go
+++ b/stream/iterator_test.go
@@ -140,7 +140,7 @@ func TestPkScan(t *testing.T) {
 			testutil.MakeDocuments(t, `{"a": 1}`, `{"a": 2}`),
 			testutil.MakeDocuments(t, `{"a": 1}`, `{"a": 2}`),
 			stream.Ranges{
-				{Max: parser.MustParseExpr("2")},
+				{Max: document.NewIntegerValue(2)},
 			},
 			false, false,
 		},
@@ -149,7 +149,7 @@ func TestPkScan(t *testing.T) {
 			testutil.MakeDocuments(t, `{"a": 1}`, `{"a": 2}`),
 			testutil.MakeDocuments(t, `{"a": 1}`),
 			stream.Ranges{
-				{Max: parser.MustParseExpr("1")},
+				{Max: document.NewIntegerValue(1)},
 			},
 			false, false,
 		},
@@ -158,7 +158,7 @@ func TestPkScan(t *testing.T) {
 			testutil.MakeDocuments(t, `{"a": 1}`, `{"a": 2}`),
 			testutil.MakeDocuments(t, `{"a": 1}`, `{"a": 2}`),
 			stream.Ranges{
-				{Min: parser.MustParseExpr("1")},
+				{Min: document.NewIntegerValue(1)},
 			},
 			false, false,
 		},
@@ -167,7 +167,7 @@ func TestPkScan(t *testing.T) {
 			testutil.MakeDocuments(t, `{"a": 1}`, `{"a": 2}`),
 			testutil.MakeDocuments(t, `{"a": 1}`, `{"a": 2}`),
 			stream.Ranges{
-				{Min: parser.MustParseExpr("1"), Max: parser.MustParseExpr("2")},
+				{Min: document.NewIntegerValue(1), Max: document.NewIntegerValue(2)},
 			},
 			false, false,
 		},
@@ -182,7 +182,7 @@ func TestPkScan(t *testing.T) {
 			testutil.MakeDocuments(t, `{"a": 1}`, `{"a": 2}`),
 			testutil.MakeDocuments(t, `{"a": 2}`, `{"a": 1}`),
 			stream.Ranges{
-				{Max: parser.MustParseExpr("2")},
+				{Max: document.NewIntegerValue(2)},
 			},
 			true, false,
 		},
@@ -191,7 +191,7 @@ func TestPkScan(t *testing.T) {
 			testutil.MakeDocuments(t, `{"a": 1}`, `{"a": 2}`),
 			testutil.MakeDocuments(t, `{"a": 2}`, `{"a": 1}`),
 			stream.Ranges{
-				{Min: parser.MustParseExpr("1")},
+				{Min: document.NewIntegerValue(1)},
 			},
 			true, false,
 		},
@@ -200,7 +200,7 @@ func TestPkScan(t *testing.T) {
 			testutil.MakeDocuments(t, `{"a": 1}`, `{"a": 2}`),
 			testutil.MakeDocuments(t, `{"a": 2}`, `{"a": 1}`),
 			stream.Ranges{
-				{Min: parser.MustParseExpr("1"), Max: parser.MustParseExpr("2")},
+				{Min: document.NewIntegerValue(1), Max: document.NewIntegerValue(2)},
 			},
 			true, false,
 		},
@@ -259,13 +259,13 @@ func TestPkScan(t *testing.T) {
 
 	t.Run("String", func(t *testing.T) {
 		require.Equal(t, `pkScan("test", [1, 2])`, stream.PkScan("test", stream.Range{
-			Min: parser.MustParseExpr("1"), Max: parser.MustParseExpr("2"),
+			Min: document.NewIntegerValue(1), Max: document.NewIntegerValue(2),
 		}).String())
 
 		op := stream.PkScan("test",
-			stream.Range{Min: parser.MustParseExpr("1"), Max: parser.MustParseExpr("2"), Exclusive: true},
-			stream.Range{Min: parser.MustParseExpr("10"), Exact: true},
-			stream.Range{Min: parser.MustParseExpr("100")},
+			stream.Range{Min: document.NewIntegerValue(1), Max: document.NewIntegerValue(2), Exclusive: true},
+			stream.Range{Min: document.NewIntegerValue(10), Exact: true},
+			stream.Range{Min: document.NewIntegerValue(100)},
 		)
 		op.Reverse = true
 
@@ -293,7 +293,7 @@ func TestIndexScan(t *testing.T) {
 			testutil.MakeDocuments(t, `{"a": 1}`, `{"a": 2}`),
 			testutil.MakeDocuments(t, `{"a": 1}`, `{"a": 2}`),
 			stream.Ranges{
-				{Max: parser.MustParseExpr("2")},
+				{Max: document.NewIntegerValue(2)},
 			},
 			false, false,
 		},
@@ -302,7 +302,7 @@ func TestIndexScan(t *testing.T) {
 			testutil.MakeDocuments(t, `{"a": 1}`, `{"a": 2}`),
 			testutil.MakeDocuments(t, `{"a": 1}`),
 			stream.Ranges{
-				{Max: parser.MustParseExpr("1")},
+				{Max: document.NewIntegerValue(1)},
 			},
 			false, false,
 		},
@@ -311,7 +311,7 @@ func TestIndexScan(t *testing.T) {
 			testutil.MakeDocuments(t, `{"a": 1}`, `{"a": 2}`),
 			testutil.MakeDocuments(t, `{"a": 1}`, `{"a": 2}`),
 			stream.Ranges{
-				{Min: parser.MustParseExpr("1")},
+				{Min: document.NewIntegerValue(1)},
 			},
 			false, false,
 		},
@@ -320,7 +320,7 @@ func TestIndexScan(t *testing.T) {
 			testutil.MakeDocuments(t, `{"a": 1}`, `{"a": 2}`),
 			testutil.MakeDocuments(t, `{"a": 1}`, `{"a": 2}`),
 			stream.Ranges{
-				{Min: parser.MustParseExpr("1"), Max: parser.MustParseExpr("2")},
+				{Min: document.NewIntegerValue(1), Max: document.NewIntegerValue(2)},
 			},
 			false, false,
 		},
@@ -335,7 +335,7 @@ func TestIndexScan(t *testing.T) {
 			testutil.MakeDocuments(t, `{"a": 1}`, `{"a": 2}`),
 			testutil.MakeDocuments(t, `{"a": 2}`, `{"a": 1}`),
 			stream.Ranges{
-				{Max: parser.MustParseExpr("2")},
+				{Max: document.NewIntegerValue(2)},
 			},
 			true, false,
 		},
@@ -344,7 +344,7 @@ func TestIndexScan(t *testing.T) {
 			testutil.MakeDocuments(t, `{"a": 1}`, `{"a": 2}`),
 			testutil.MakeDocuments(t, `{"a": 2}`, `{"a": 1}`),
 			stream.Ranges{
-				{Min: parser.MustParseExpr("1")},
+				{Min: document.NewIntegerValue(1)},
 			},
 			true, false,
 		},
@@ -353,7 +353,7 @@ func TestIndexScan(t *testing.T) {
 			testutil.MakeDocuments(t, `{"a": 1}`, `{"a": 2}`),
 			testutil.MakeDocuments(t, `{"a": 2}`, `{"a": 1}`),
 			stream.Ranges{
-				{Min: parser.MustParseExpr("1"), Max: parser.MustParseExpr("2")},
+				{Min: document.NewIntegerValue(1), Max: document.NewIntegerValue(2)},
 			},
 			true, false,
 		},
@@ -412,11 +412,11 @@ func TestIndexScan(t *testing.T) {
 
 	t.Run("String", func(t *testing.T) {
 		require.Equal(t, `indexScan("idx_test_a", [1, 2])`, stream.IndexScan("idx_test_a", stream.Range{
-			Min: parser.MustParseExpr("1"), Max: parser.MustParseExpr("2"),
+			Min: document.NewIntegerValue(1), Max: document.NewIntegerValue(2),
 		}).String())
 
 		op := stream.IndexScan("idx_test_a", stream.Range{
-			Min: parser.MustParseExpr("1"), Max: parser.MustParseExpr("2"),
+			Min: document.NewIntegerValue(1), Max: document.NewIntegerValue(2),
 		})
 		op.Reverse = true