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) + } + } + }) +}