From 2b88d278bc3d138df64275e7794c916fc83f032c Mon Sep 17 00:00:00 2001
From: Travis <travis@pilosa.com>
Date: Mon, 18 Sep 2017 11:59:44 -0500
Subject: [PATCH 1/3] Implements BETWEEN for Range queries.

The PQL looks like:
```
Range(frame=f, field0 >< [200,610])
```

One thing I noticed while implementing this is that it doesn't seem
like `FieldRange()` is used in either `Frame` or `View`; the Executor
calls `Fragment.FieldRange()` directly. The problem with this is that
the offset logic is calculated in the Frame, but since the Executor
doesn't go through Frame, then the Executor also has to calculate
the offset before calling `Fragment.FieldRange`. We should unify this
logic somewhere. Note, this applies to both `FieldRange` and
`FieldRangeBetween`.
---
 executor.go         | 100 +++++++++++++++++++++++++++++++++++---------
 fragment.go         |  53 ++++++++++++++++++++++-
 fragment_test.go    |  48 +++++++++++++++++++++
 frame.go            |  52 ++++++++++++++++++++++-
 pilosa.go           |   1 +
 pql/ast.go          |  25 +++++++++++
 pql/ast_test.go     |  26 ++++++++++++
 pql/parser.go       |   2 +-
 pql/parser_test.go  |   3 +-
 pql/scanner.go      |   5 ++-
 pql/scanner_test.go |   1 +
 pql/token.go        |  46 ++++++++++----------
 view.go             |  14 +++++++
 13 files changed, 329 insertions(+), 47 deletions(-)

diff --git a/executor.go b/executor.go
index 1507c3050..3d07ebbb8 100644
--- a/executor.go
+++ b/executor.go
@@ -715,28 +715,90 @@ func (e *Executor) executeFieldRangeSlice(ctx context.Context, index string, c *
 		fieldName, cond = k, vv
 	}
 
-	// Only support integers for now.
-	value, ok := cond.Value.(int64)
-	if !ok {
-		return nil, errors.New("Range(): conditions only support integer values")
-	}
+	if cond.Op == pql.BETWEEN {
 
-	// Find field.
-	field := f.Field(fieldName)
-	if field == nil {
-		return nil, ErrFieldNotFound
-	} else if value < field.Min || value > field.Max {
-		return NewBitmap(), nil
-	}
+		predicates, err := cond.IntSliceValue()
+		if err != nil {
+			return nil, err
+		}
 
-	// Retrieve fragment.
-	frag := e.Holder.Fragment(index, frame, ViewFieldPrefix+fieldName, slice)
-	if frag == nil {
-		return NewBitmap(), nil
-	}
+		// Only support two integers for the between operation.
+		if len(predicates) != 2 {
+			return nil, errors.New("Range(): BETWEEN condition requires exactly two integer values")
+		}
 
-	f.Stats.Count("range:field", 1, 1.0)
-	return frag.FieldRange(cond.Op, field.BitDepth(), uint64(value-field.Min))
+		// Find field.
+		field := f.Field(fieldName)
+		if field == nil {
+			return nil, ErrFieldNotFound
+		} else if predicates[1] < field.Min || predicates[0] > field.Max {
+			return NewBitmap(), nil
+		}
+
+		// Adjust predicates to range.
+		baseValueMin := uint64(0)
+		baseValueMax := uint64(0)
+		if predicates[0] > field.Min {
+			baseValueMin = uint64(predicates[0] - field.Min)
+		}
+		// Make sure the high value in our BETWEEN does not exceed BitDepth.
+		if predicates[1] > field.Max {
+			baseValueMax = uint64(field.Max - field.Min)
+		} else if predicates[1] > field.Min {
+			baseValueMax = uint64(predicates[1] - field.Min)
+		}
+
+		// Retrieve fragment.
+		frag := e.Holder.Fragment(index, frame, ViewFieldPrefix+fieldName, slice)
+		if frag == nil {
+			return NewBitmap(), nil
+		}
+
+		return frag.FieldRangeBetween(field.BitDepth(), baseValueMin, baseValueMax)
+
+	} else {
+
+		// Only support integers for now.
+		value, ok := cond.Value.(int64)
+		if !ok {
+			return nil, errors.New("Range(): conditions only support integer values")
+		}
+
+		// Find field.
+		field := f.Field(fieldName)
+		if field == nil {
+			return nil, ErrFieldNotFound
+		}
+
+		// Adjust predicate to range.
+		baseValue := uint64(0)
+		if cond.Op == pql.GT || cond.Op == pql.GTE {
+			if value > field.Max {
+				return NewBitmap(), nil
+			} else if value > field.Min {
+				baseValue = uint64(value - field.Min)
+			}
+		} else if cond.Op == pql.LT || cond.Op == pql.LTE {
+			if value < field.Min {
+				return NewBitmap(), nil
+			} else if value > field.Max {
+				baseValue = uint64(field.Max - field.Min)
+			} else {
+				baseValue = uint64(value - field.Min)
+			}
+		} else if cond.Op == pql.EQ {
+			baseValue = uint64(value - field.Min)
+		}
+
+		// Retrieve fragment.
+		frag := e.Holder.Fragment(index, frame, ViewFieldPrefix+fieldName, slice)
+		if frag == nil {
+			return NewBitmap(), nil
+		}
+
+		f.Stats.Count("range:field", 1, 1.0)
+		return frag.FieldRange(cond.Op, field.BitDepth(), baseValue)
+	}
 }
 
 // executeUnionSlice executes a union() call for a local slice.
diff --git a/fragment.go b/fragment.go
index 4fa13e70e..b83dd579d 100644
--- a/fragment.go
+++ b/fragment.go
@@ -644,7 +644,10 @@ func (f *Fragment) fieldRangeLT(bitDepth uint, predicate uint64, allowEquality b
 		}
 
 		// If bit is set then add columns for set bits to exclude.
-		keep = keep.Union(b.Difference(row))
+		// Don't bother to compute this on the final iteration.
+		if i > 0 {
+			keep = keep.Union(b.Difference(row))
+		}
 	}
 
 	return b, nil
@@ -676,7 +679,53 @@ func (f *Fragment) fieldRangeGT(bitDepth uint, predicate uint64, allowEquality b
 		}
 
 		// If bit is unset then add columns with set bit to keep.
-		keep = keep.Union(b.Intersect(row))
+		// Don't bother to compute this on the final iteration.
+		if i > 0 {
+			keep = keep.Union(b.Intersect(row))
+		}
+	}
+
+	return b, nil
+}
+
+func (f *Fragment) FieldRangeBetween(bitDepth uint, predicateMin, predicateMax uint64) (*Bitmap, error) {
+	return f.fieldRangeBetween(bitDepth, predicateMin, predicateMax)
+}
+
+func (f *Fragment) fieldRangeBetween(bitDepth uint, predicateMin, predicateMax uint64) (*Bitmap, error) {
+	b := f.Row(uint64(bitDepth))
+	keep1 := NewBitmap() // GTE
+	keep2 := NewBitmap() // LTE
+
+	// Filter any bits that don't match the current bit value.
+	for i := int(bitDepth - 1); i >= 0; i-- {
+		row := f.Row(uint64(i))
+		bit1 := (predicateMin >> uint(i)) & 1
+		bit2 := (predicateMax >> uint(i)) & 1
+
+		// GTE predicateMin
+		// If bit is set then remove all unset columns not already kept.
+		if bit1 == 1 {
+			b = b.Difference(b.Difference(row).Difference(keep1))
+		} else {
+			// If bit is unset then add columns with set bit to keep.
+			// Don't bother to compute this on the final iteration.
+			if i > 0 {
+				keep1 = keep1.Union(b.Intersect(row))
+			}
+		}
+
+		// LTE predicateMin
+		// If bit is zero then remove all set columns not in excluded bitmap.
+		if bit2 == 0 {
+			b = b.Difference(row.Difference(keep2))
+		} else {
+			// If bit is set then add columns for set bits to exclude.
+			// Don't bother to compute this on the final iteration.
+			if i > 0 {
+				keep2 = keep2.Union(b.Difference(row))
+			}
+		}
 	}
 
 	return b, nil
diff --git a/fragment_test.go b/fragment_test.go
index 941554db4..21723dc3b 100644
--- a/fragment_test.go
+++ b/fragment_test.go
@@ -378,6 +378,54 @@ func TestFragment_FieldRange(t *testing.T) {
 			t.Fatalf("unexpected bits: %+v", b.Bits())
 		}
 	})
+
+	t.Run("BETWEEN", func(t *testing.T) {
+		f := test.MustOpenFragment("i", "f", pilosa.ViewStandard, 0, "")
+		defer f.Close()
+
+		// Set values.
+		if _, err := f.SetFieldValue(1000, bitDepth, 382); err != nil {
+			t.Fatal(err)
+		} else if _, err := f.SetFieldValue(2000, bitDepth, 300); err != nil {
+			t.Fatal(err)
+		} else if _, err := f.SetFieldValue(3000, bitDepth, 2817); err != nil {
+			t.Fatal(err)
+		} else if _, err := f.SetFieldValue(4000, bitDepth, 301); err != nil {
+			t.Fatal(err)
+		} else if _, err := f.SetFieldValue(5000, bitDepth, 1); err != nil {
+			t.Fatal(err)
+		} else if _, err := f.SetFieldValue(6000, bitDepth, 0); err != nil {
+			t.Fatal(err)
+		}
+
+		// Query for fields greater than (ending with unset bit).
+		if b, err := f.FieldRangeBetween(bitDepth, 300, 2817); err != nil {
+			t.Fatal(err)
+		} else if !reflect.DeepEqual(b.Bits(), []uint64{1000, 2000, 3000, 4000}) {
+			t.Fatalf("unexpected bits: %+v", b.Bits())
+		}
+
+		// Query for fields greater than (ending with set bit).
+		if b, err := f.FieldRangeBetween(bitDepth, 301, 2817); err != nil {
+			t.Fatal(err)
+		} else if !reflect.DeepEqual(b.Bits(), []uint64{1000, 3000, 4000}) {
+			t.Fatalf("unexpected bits: %+v", b.Bits())
+		}
+
+		// Query for fields greater than or equal to (ending with unset bit).
+		if b, err := f.FieldRangeBetween(bitDepth, 301, 2816); err != nil {
+			t.Fatal(err)
+		} else if !reflect.DeepEqual(b.Bits(), []uint64{1000, 4000}) {
+			t.Fatalf("unexpected bits: %+v", b.Bits())
+		}
+
+		// Query for fields greater than or equal to (ending with set bit).
+		if b, err := f.FieldRangeBetween(bitDepth, 300, 2816); err != nil {
+			t.Fatal(err)
+		} else if !reflect.DeepEqual(b.Bits(), []uint64{1000, 2000, 4000}) {
+			t.Fatalf("unexpected bits: %+v", b.Bits())
+		}
+	})
 }
 
 // Ensure a fragment can snapshot correctly.
diff --git a/frame.go b/frame.go
index f9580cc6b..498b6858b 100644
--- a/frame.go
+++ b/frame.go
@@ -747,11 +747,61 @@ func (f *Frame) FieldRange(name string, op pql.Token, predicate int64) (*Bitmap,
 	}
 
 	// Adjust predicate to range.
-	baseValue := uint64(predicate - field.Min)
+	baseValue := uint64(0)
+	if op == pql.GT || op == pql.GTE {
+		if predicate > field.Max {
+			return NewBitmap(), nil
+		} else if predicate > field.Min {
+			baseValue = uint64(predicate - field.Min)
+		}
+	} else if op == pql.LT || op == pql.LTE {
+		if predicate < field.Min {
+			return NewBitmap(), nil
+		} else if predicate > field.Max {
+			baseValue = uint64(field.Max - field.Min)
+		} else {
+			baseValue = uint64(predicate - field.Min)
+		}
+	} else if op == pql.EQ {
+		baseValue = uint64(predicate - field.Min)
+	}
 
 	return view.FieldRange(op, field.BitDepth(), baseValue)
 }
 
+func (f *Frame) FieldRangeBetween(name string, predicateMin, predicateMax int64) (*Bitmap, error) {
+	// Retrieve and validate field.
+	field := f.Field(name)
+	if field == nil {
+		return nil, ErrFieldNotFound
+	} else if predicateMin > predicateMax {
+		return nil, ErrInvalidBetweenValue
+	} else if predicateMax < field.Min || predicateMin > field.Max {
+		return nil, nil
+	}
+
+	// Retrieve field's view.
+	view := f.View(ViewFieldPrefix + name)
+	if view == nil {
+		return nil, nil
+	}
+
+	// Adjust predicates to range.
+	baseValueMin := uint64(0)
+	baseValueMax := uint64(0)
+	if predicateMin > field.Min {
+		baseValueMin = uint64(predicateMin - field.Min)
+	}
+	// Make sure the high value in our BETWEEN does not exceed BitDepth.
+	if predicateMax > field.Max {
+		baseValueMax = uint64(field.Max - field.Min)
+	} else if predicateMax > field.Min {
+		baseValueMax = uint64(predicateMax - field.Min)
+	}
+
+	return view.FieldRangeBetween(field.BitDepth(), baseValueMin, baseValueMax)
+}
+
 // Import bulk imports data.
 func (f *Frame) Import(rowIDs, columnIDs []uint64, timestamps []*time.Time) error {
 	// Determine quantum if timestamps are set.
diff --git a/pilosa.go b/pilosa.go
index 49bbd6c11..dfa8a615f 100644
--- a/pilosa.go
+++ b/pilosa.go
@@ -60,6 +60,7 @@ var (
 	ErrFieldValueTooLow       = errors.New("field value too low")
 	ErrFieldValueTooHigh      = errors.New("field value too high")
 	ErrInvalidRangeOperation  = errors.New("invalid range operation")
+	ErrInvalidBetweenValue    = errors.New("invalid value for between operation")
 
 	ErrInvalidView      = errors.New("invalid view")
 	ErrInvalidCacheType = errors.New("invalid cache type")
diff --git a/pql/ast.go b/pql/ast.go
index 48e5f1771..e01540b84 100644
--- a/pql/ast.go
+++ b/pql/ast.go
@@ -220,6 +220,31 @@ func (cond *Condition) String() string {
 	return fmt.Sprintf("%s %s", cond.Op.String(), FormatValue(cond.Value))
 }
 
+// IntSliceValue reads cond.Value as a slice of uint64.
+// If the value is a slice of uint64 it will convert
+// it to []int64. Otherwise, if it is not a []int64 it will return an error.
+func (cond *Condition) IntSliceValue() ([]int64, error) {
+	val := cond.Value
+
+	switch tval := val.(type) {
+	case []interface{}:
+		ret := make([]int64, len(tval))
+		for i, v := range tval {
+			switch tv := v.(type) {
+			case int64:
+				ret[i] = tv
+			case uint64:
+				ret[i] = int64(tv)
+			default:
+				return nil, fmt.Errorf("unexpected value type %T in IntSliceValue, val %v", tv, tv)
+			}
+		}
+		return ret, nil
+	default:
+		return nil, fmt.Errorf("unexpected type %T in IntSliceValue, val %v", tval, tval)
+	}
+}
+
 func FormatValue(v interface{}) string {
 	switch v := v.(type) {
 	case string:
diff --git a/pql/ast_test.go b/pql/ast_test.go
index 2b8519571..b8416eccb 100644
--- a/pql/ast_test.go
+++ b/pql/ast_test.go
@@ -15,6 +15,7 @@
 package pql_test
 
 import (
+	"reflect"
 	"testing"
 
 	"github.com/pilosa/pilosa/pql"
@@ -30,6 +31,31 @@ func TestCall_String(t *testing.T) {
 	})
 }
 
+// Ensure condition can handle values for BETWEEN operator.
+func TestCondition_Value(t *testing.T) {
+	t.Run("Between Values", func(t *testing.T) {
+		for _, tt := range []struct {
+			val []interface{}
+			exp []int64
+		}{
+			{[]interface{}{int64(4), int64(8)}, []int64{4, 8}},
+			{[]interface{}{uint64(4), uint64(8)}, []int64{4, 8}},
+			{[]interface{}{uint64(1), uint64(2), uint64(3)}, []int64{1, 2, 3}},
+		} {
+			c := &pql.Condition{
+				Op:    pql.BETWEEN,
+				Value: tt.val,
+			}
+			v, err := c.IntSliceValue()
+			if err != nil {
+				t.Fatal(err)
+			} else if !reflect.DeepEqual(v, tt.exp) {
+				t.Fatalf("invalid between values. expected: %v, got %v", tt.exp, v)
+			}
+		}
+	})
+}
+
 // Ensure call can be converted into a string.
 func TestCall_SupportsInverse(t *testing.T) {
 	t.Run("Bitmap", func(t *testing.T) {
diff --git a/pql/parser.go b/pql/parser.go
index f0d5feeae..ab5298458 100644
--- a/pql/parser.go
+++ b/pql/parser.go
@@ -165,7 +165,7 @@ func (p *Parser) parseArgs() (map[string]interface{}, error) {
 		var op Token
 		switch tok, pos, lit := p.scanIgnoreWhitespace(); tok {
 		case ASSIGN:
-		case EQ, LT, LTE, GT, GTE:
+		case EQ, LT, LTE, GT, GTE, BETWEEN:
 			op = tok
 		default:
 			return nil, parseErrorf(pos, "expected equals sign or comparison operator, found %q", lit)
diff --git a/pql/parser_test.go b/pql/parser_test.go
index 736a9c7f8..266e1de1f 100644
--- a/pql/parser_test.go
+++ b/pql/parser_test.go
@@ -172,7 +172,7 @@ func TestParser_Parse(t *testing.T) {
 
 	// Parse with condition arguments.
 	t.Run("WithCondition", func(t *testing.T) {
-		q, err := pql.ParseString(`MyCall(key=foo, x == 12.25, y >= 100)`)
+		q, err := pql.ParseString(`MyCall(key=foo, x == 12.25, y >= 100, z >< [4,8])`)
 		if err != nil {
 			t.Fatal(err)
 		} else if !reflect.DeepEqual(q.Calls[0],
@@ -182,6 +182,7 @@ func TestParser_Parse(t *testing.T) {
 					"key": "foo",
 					"x":   &pql.Condition{Op: pql.EQ, Value: 12.25},
 					"y":   &pql.Condition{Op: pql.GTE, Value: int64(100)},
+					"z":   &pql.Condition{Op: pql.BETWEEN, Value: []interface{}{int64(4), int64(8)}},
 				},
 			},
 		) {
diff --git a/pql/scanner.go b/pql/scanner.go
index 38fd7d6fc..f25abf1f4 100644
--- a/pql/scanner.go
+++ b/pql/scanner.go
@@ -73,8 +73,11 @@ func (s *Scanner) Scan() (tok Token, pos Pos, lit string) {
 		s.unread()
 		return LT, pos, string(ch)
 	case '>':
-		if next := s.read(); next == '=' {
+		next := s.read()
+		if next == '=' {
 			return GTE, pos, ">="
+		} else if next == '<' {
+			return BETWEEN, pos, "><"
 		}
 		s.unread()
 		return GT, pos, string(ch)
diff --git a/pql/scanner_test.go b/pql/scanner_test.go
index 49d272ce5..98c13e232 100644
--- a/pql/scanner_test.go
+++ b/pql/scanner_test.go
@@ -42,6 +42,7 @@ func TestScanner_Scan(t *testing.T) {
 		{name: "LTE", s: `<=`, tok: pql.LTE, lit: `<=`},
 		{name: "GT", s: `>`, tok: pql.GT, lit: `>`},
 		{name: "GTE", s: `>=`, tok: pql.GTE, lit: `>=`},
+		{name: "BETWEEN", s: `><`, tok: pql.BETWEEN, lit: `><`},
 		{name: "COMMA", s: `,`, tok: pql.COMMA, lit: `,`},
 		{name: "LPAREN", s: `(`, tok: pql.LPAREN, lit: `(`},
 		{name: "RPAREN", s: `)`, tok: pql.RPAREN, lit: `)`},
diff --git a/pql/token.go b/pql/token.go
index 1327df9ce..2870553e4 100644
--- a/pql/token.go
+++ b/pql/token.go
@@ -37,17 +37,18 @@ const (
 	ALL
 	keyword_end
 
-	ASSIGN // =
-	EQ     // ==
-	LT     // <
-	LTE    // <=
-	GT     // >
-	GTE    // >=
-	COMMA  // ,
-	LPAREN // (
-	RPAREN // )
-	LBRACK // (
-	RBRACK // )
+	ASSIGN  // =
+	EQ      // ==
+	LT      // <
+	LTE     // <=
+	GT      // >
+	GTE     // >=
+	BETWEEN // ><
+	COMMA   // ,
+	LPAREN  // (
+	RPAREN  // )
+	LBRACK  // (
+	RBRACK  // )
 )
 
 var tokens = [...]string{
@@ -61,17 +62,18 @@ var tokens = [...]string{
 
 	ALL: "ALL",
 
-	ASSIGN: "=",
-	EQ:     "==",
-	LT:     "<",
-	LTE:    "<=",
-	GT:     ">",
-	GTE:    ">=",
-	COMMA:  ",",
-	LPAREN: "(",
-	RPAREN: ")",
-	LBRACK: "(",
-	RBRACK: ")",
+	ASSIGN:  "=",
+	EQ:      "==",
+	LT:      "<",
+	LTE:     "<=",
+	GT:      ">",
+	GTE:     ">=",
+	BETWEEN: "><",
+	COMMA:   ",",
+	LPAREN:  "(",
+	RPAREN:  ")",
+	LBRACK:  "(",
+	RBRACK:  ")",
 }
 
 var keywords map[string]Token
diff --git a/view.go b/view.go
index 2d5b7d776..7a8ed4439 100644
--- a/view.go
+++ b/view.go
@@ -329,6 +329,20 @@ func (v *View) FieldRange(op pql.Token, bitDepth uint, predicate uint64) (*Bitma
 	return bm, nil
 }
 
+// FieldRangeBetween returns bitmaps with a field value encoding matching any
+// value between predicateMin and predicateMax.
+func (v *View) FieldRangeBetween(bitDepth uint, predicateMin, predicateMax uint64) (*Bitmap, error) {
+	bm := NewBitmap()
+	for _, frag := range v.Fragments() {
+		other, err := frag.FieldRangeBetween(bitDepth, predicateMin, predicateMax)
+		if err != nil {
+			return nil, err
+		}
+		bm = bm.Union(other)
+	}
+	return bm, nil
+}
+
 // IsInverseView returns true if the view is used for storing an inverted representation.
 func IsInverseView(name string) bool {
 	return strings.HasPrefix(name, ViewInverse)

From e53f218c49c84f1b18d5ea4edd0714ac4dae1d15 Mon Sep 17 00:00:00 2001
From: Travis <travis@pilosa.com>
Date: Tue, 19 Sep 2017 17:17:45 -0500
Subject: [PATCH 2/3] consolidate the BaseValue code. Add tests for BaseValue
 ranges

---
 executor.go   |  37 +++------------
 frame.go      |  89 ++++++++++++++++++++++------------
 frame_test.go | 129 ++++++++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 193 insertions(+), 62 deletions(-)

diff --git a/executor.go b/executor.go
index 3d07ebbb8..3f332bd1a 100644
--- a/executor.go
+++ b/executor.go
@@ -731,21 +731,11 @@ func (e *Executor) executeFieldRangeSlice(ctx context.Context, index string, c *
 		field := f.Field(fieldName)
 		if field == nil {
 			return nil, ErrFieldNotFound
-		} else if predicates[1] < field.Min || predicates[0] > field.Max {
-			return NewBitmap(), nil
 		}
 
-		// Adjust predicates to range.
-		baseValueMin := uint64(0)
-		baseValueMax := uint64(0)
-		if predicates[0] > field.Min {
-			baseValueMin = uint64(predicates[0] - field.Min)
-		}
-		// Make sure the high value in our BETWEEN does not exceed BitDepth.
-		if predicates[1] > field.Max {
-			baseValueMax = uint64(field.Max - field.Min)
-		} else if predicates[1] > field.Min {
-			baseValueMax = uint64(predicates[1] - field.Min)
+		baseValueMin, baseValueMax, outOfRange := field.BaseValueBetween(predicates[0], predicates[1])
+		if outOfRange {
+			return NewBitmap(), nil
 		}
 
 		// Retrieve fragment.
@@ -770,24 +760,9 @@ func (e *Executor) executeFieldRangeSlice(ctx context.Context, index string, c *
 			return nil, ErrFieldNotFound
 		}
 
-		// Adjust predicate to range.
-		baseValue := uint64(0)
-		if cond.Op == pql.GT || cond.Op == pql.GTE {
-			if value > field.Max {
-				return NewBitmap(), nil
-			} else if value > field.Min {
-				baseValue = uint64(value - field.Min)
-			}
-		} else if cond.Op == pql.LT || cond.Op == pql.LTE {
-			if value < field.Min {
-				return NewBitmap(), nil
-			} else if value > field.Max {
-				baseValue = uint64(field.Max - field.Min)
-			} else {
-				baseValue = uint64(value - field.Min)
-			}
-		} else if cond.Op == pql.EQ {
-			baseValue = uint64(value - field.Min)
+		baseValue, outOfRange := field.BaseValue(cond.Op, value)
+		if outOfRange {
+			return NewBitmap(), nil
 		}
 
 		// Retrieve fragment.
diff --git a/frame.go b/frame.go
index 498b6858b..a7cf82812 100644
--- a/frame.go
+++ b/frame.go
@@ -746,24 +746,9 @@ func (f *Frame) FieldRange(name string, op pql.Token, predicate int64) (*Bitmap,
 		return nil, nil
 	}
 
-	// Adjust predicate to range.
-	baseValue := uint64(0)
-	if op == pql.GT || op == pql.GTE {
-		if predicate > field.Max {
-			return NewBitmap(), nil
-		} else if predicate > field.Min {
-			baseValue = uint64(predicate - field.Min)
-		}
-	} else if op == pql.LT || op == pql.LTE {
-		if predicate < field.Min {
-			return NewBitmap(), nil
-		} else if predicate > field.Max {
-			baseValue = uint64(field.Max - field.Min)
-		} else {
-			baseValue = uint64(predicate - field.Min)
-		}
-	} else if op == pql.EQ {
-		baseValue = uint64(predicate - field.Min)
+	baseValue, outOfRange := field.BaseValue(op, predicate)
+	if outOfRange {
+		return NewBitmap(), nil
 	}
 
 	return view.FieldRange(op, field.BitDepth(), baseValue)
@@ -776,8 +761,6 @@ func (f *Frame) FieldRangeBetween(name string, predicateMin, predicateMax int64)
 		return nil, ErrFieldNotFound
 	} else if predicateMin > predicateMax {
 		return nil, ErrInvalidBetweenValue
-	} else if predicateMax < field.Min || predicateMin > field.Max {
-		return nil, nil
 	}
 
 	// Retrieve field's view.
@@ -786,17 +769,9 @@ func (f *Frame) FieldRangeBetween(name string, predicateMin, predicateMax int64)
 		return nil, nil
 	}
 
-	// Adjust predicates to range.
-	baseValueMin := uint64(0)
-	baseValueMax := uint64(0)
-	if predicateMin > field.Min {
-		baseValueMin = uint64(predicateMin - field.Min)
-	}
-	// Make sure the high value in our BETWEEN does not exceed BitDepth.
-	if predicateMax > field.Max {
-		baseValueMax = uint64(field.Max - field.Min)
-	} else if predicateMax > field.Min {
-		baseValueMax = uint64(predicateMax - field.Min)
+	baseValueMin, baseValueMax, outOfRange := field.BaseValueBetween(predicateMin, predicateMax)
+	if outOfRange {
+		return NewBitmap(), nil
 	}
 
 	return view.FieldRangeBetween(field.BitDepth(), baseValueMin, baseValueMax)
@@ -1049,6 +1024,58 @@ func (f *Field) BitDepth() uint {
 	return 63
 }
 
+// BaseValue adjusts the value to align with the range for Field for a certain
+// operation type.
+// TODO: there is an edge case for GT and LT where this returns a baseValue
+// that does not fully encompass the range.
+// ex: Field.Min = 0, Field.Max = 1023
+// BaseValue(LT, 2000) returns 1023, which will perform "LT 1023" and effectively
+// exclude any columns with value = 1023.
+// Note that in this case (because the range uses the full BitDepth 0 to 1023),
+// we can't simply return 1024.
+// In order to make this work, we effectively need to change the operator to LTE.
+func (f *Field) BaseValue(op pql.Token, value int64) (baseValue uint64, outOfRange bool) {
+	if op == pql.GT || op == pql.GTE {
+		if value > f.Max {
+			return baseValue, true
+		} else if value > f.Min {
+			baseValue = uint64(value - f.Min)
+		}
+	} else if op == pql.LT || op == pql.LTE {
+		if value < f.Min {
+			return baseValue, true
+		} else if value > f.Max {
+			baseValue = uint64(f.Max - f.Min)
+		} else {
+			baseValue = uint64(value - f.Min)
+		}
+	} else if op == pql.EQ {
+		if value < f.Min || value > f.Max {
+			return baseValue, true
+		}
+		baseValue = uint64(value - f.Min)
+	}
+	return baseValue, false
+}
+
+// BaseValueBetween adjusts the min/max value to align with the range for Field.
+func (f *Field) BaseValueBetween(min, max int64) (baseValueMin, baseValueMax uint64, outOfRange bool) {
+	if max < f.Min || min > f.Max {
+		return baseValueMin, baseValueMax, true
+	}
+	// Adjust min/max to range.
+	if min > f.Min {
+		baseValueMin = uint64(min - f.Min)
+	}
+	// Make sure the high value of the BETWEEN does not exceed BitDepth.
+	if max > f.Max {
+		baseValueMax = uint64(f.Max - f.Min)
+	} else if max > f.Min {
+		baseValueMax = uint64(max - f.Min)
+	}
+	return baseValueMin, baseValueMax, false
+}
+
 func ValidateField(f *Field) error {
 	if f.Name == "" {
 		return ErrFieldNameRequired
diff --git a/frame_test.go b/frame_test.go
index 50f5fdef5..6e4ade84a 100644
--- a/frame_test.go
+++ b/frame_test.go
@@ -16,9 +16,11 @@ package pilosa_test
 
 import (
 	"io/ioutil"
+	"reflect"
 	"testing"
 
 	"github.com/pilosa/pilosa"
+	"github.com/pilosa/pilosa/pql"
 	"github.com/pilosa/pilosa/test"
 )
 
@@ -340,3 +342,130 @@ func TestFrame_DeleteView(t *testing.T) {
 		t.Fatal("failed to create new view")
 	}
 }
+
+// Ensure a field can adjust to its baseValue.
+func TestField_BaseValue(t *testing.T) {
+	f0 := &pilosa.Field{
+		Name: "f0",
+		Type: pilosa.FieldTypeInt,
+		Min:  -100,
+		Max:  900,
+	}
+	f1 := &pilosa.Field{
+		Name: "f1",
+		Type: pilosa.FieldTypeInt,
+		Min:  0,
+		Max:  1000,
+	}
+
+	f2 := &pilosa.Field{
+		Name: "f2",
+		Type: pilosa.FieldTypeInt,
+		Min:  100,
+		Max:  1100,
+	}
+
+	t.Run("Normal Condition", func(t *testing.T) {
+
+		for _, tt := range []struct {
+			f             *pilosa.Field
+			op            pql.Token
+			val           int64
+			expBaseValue  uint64
+			expOutOfRange bool
+		}{
+			// LT
+			{f0, pql.LT, 5, 105, false},
+			{f0, pql.LT, -8, 92, false},
+			{f0, pql.LT, -108, 0, true},
+			{f0, pql.LT, 1005, 1000, false},
+			{f0, pql.LT, 0, 100, false},
+
+			{f1, pql.LT, 5, 5, false},
+			{f1, pql.LT, -8, 0, true},
+			{f1, pql.LT, 1005, 1000, false},
+			{f1, pql.LT, 0, 0, false},
+
+			{f2, pql.LT, 5, 0, true},
+			{f2, pql.LT, -8, 0, true},
+			{f2, pql.LT, 105, 5, false},
+			{f2, pql.LT, 1105, 1000, false},
+
+			// GT
+			{f0, pql.GT, -105, 0, false},
+			{f0, pql.GT, 5, 105, false},
+			{f0, pql.GT, 905, 0, true},
+			{f0, pql.GT, 0, 100, false},
+
+			{f1, pql.GT, 5, 5, false},
+			{f1, pql.GT, -8, 0, false},
+			{f1, pql.GT, 1005, 0, true},
+			{f1, pql.GT, 0, 0, false},
+
+			{f2, pql.GT, 5, 0, false},
+			{f2, pql.GT, -8, 0, false},
+			{f2, pql.GT, 105, 5, false},
+			{f2, pql.GT, 1105, 0, true},
+
+			// EQ
+			{f0, pql.EQ, -105, 0, true},
+			{f0, pql.EQ, 5, 105, false},
+			{f0, pql.EQ, 905, 0, true},
+			{f0, pql.EQ, 0, 100, false},
+
+			{f1, pql.EQ, 5, 5, false},
+			{f1, pql.EQ, -8, 0, true},
+			{f1, pql.EQ, 1005, 0, true},
+			{f1, pql.EQ, 0, 0, false},
+
+			{f2, pql.EQ, 5, 0, true},
+			{f2, pql.EQ, -8, 0, true},
+			{f2, pql.EQ, 105, 5, false},
+			{f2, pql.EQ, 1105, 0, true},
+		} {
+			bv, oor := tt.f.BaseValue(tt.op, tt.val)
+			if oor != tt.expOutOfRange {
+				t.Fatalf("baseValue calculation on %s op %s, expected outOfRange %v, got %v", tt.f.Name, tt.op, tt.expOutOfRange, oor)
+			} else if !reflect.DeepEqual(bv, tt.expBaseValue) {
+				t.Fatalf("baseValue calculation on %s, expected value %v, got %v", tt.f.Name, tt.expBaseValue, bv)
+			}
+		}
+	})
+
+	t.Run("Betwween Condition", func(t *testing.T) {
+		for _, tt := range []struct {
+			f               *pilosa.Field
+			predMin         int64
+			predMax         int64
+			expBaseValueMin uint64
+			expBaseValueMax uint64
+			expOutOfRange   bool
+		}{
+
+			{f0, -205, -105, 0, 0, true},
+			{f0, -105, 80, 0, 180, false},
+			{f0, 5, 20, 105, 120, false},
+			{f0, 20, 1005, 120, 1000, false},
+			{f0, 1005, 2000, 0, 0, true},
+
+			{f1, -105, -5, 0, 0, true},
+			{f1, -5, 20, 0, 20, false},
+			{f1, 5, 20, 5, 20, false},
+			{f1, 20, 1005, 20, 1000, false},
+			{f1, 1005, 2000, 0, 0, true},
+
+			{f2, 5, 95, 0, 0, true},
+			{f2, 95, 120, 0, 20, false},
+			{f2, 105, 120, 5, 20, false},
+			{f2, 120, 1105, 20, 1000, false},
+			{f2, 1105, 2000, 0, 0, true},
+		} {
+			min, max, oor := tt.f.BaseValueBetween(tt.predMin, tt.predMax)
+			if oor != tt.expOutOfRange {
+				t.Fatalf("baseValueBetween calculation on %s, expected outOfRange %v, got %v", tt.f.Name, tt.expOutOfRange, oor)
+			} else if !reflect.DeepEqual(min, tt.expBaseValueMin) || !reflect.DeepEqual(max, tt.expBaseValueMax) {
+				t.Fatalf("baseValueBetween calculation on %s, expected min/max %v/%v, got %v/%v", tt.f.Name, tt.expBaseValueMin, tt.expBaseValueMax, min, max)
+			}
+		}
+	})
+}

From ab908227beca13208debf67da1c901ae063903d3 Mon Sep 17 00:00:00 2001
From: Travis <travis@pilosa.com>
Date: Wed, 27 Sep 2017 11:02:32 -0500
Subject: [PATCH 3/3] remove private function fieldRangeBetween

---
 fragment.go | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/fragment.go b/fragment.go
index b83dd579d..b860f997a 100644
--- a/fragment.go
+++ b/fragment.go
@@ -689,10 +689,6 @@ func (f *Fragment) fieldRangeGT(bitDepth uint, predicate uint64, allowEquality b
 }
 
 func (f *Fragment) FieldRangeBetween(bitDepth uint, predicateMin, predicateMax uint64) (*Bitmap, error) {
-	return f.fieldRangeBetween(bitDepth, predicateMin, predicateMax)
-}
-
-func (f *Fragment) fieldRangeBetween(bitDepth uint, predicateMin, predicateMax uint64) (*Bitmap, error) {
 	b := f.Row(uint64(bitDepth))
 	keep1 := NewBitmap() // GTE
 	keep2 := NewBitmap() // LTE