scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 02:44:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1902 from benbjohnson/unbounded-bsi-sigbit

Browse source 
Unbounded BSI w/ sign magnitude
This commit is contained in:
Ben Johnson 2019-05-19 21:29:34 -06:00 • committed by GitHub
commit f59b49e4bb
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
19 changed files with 1437 additions and 3194 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
api_test.go
View file

@ -17,6 +17,7 @@ package pilosa_test
import (
"context"
"fmt"
"math"
"reflect"
"strings"
"testing"
@ -198,7 +199,7 @@ func TestAPI_ImportValue(t *testing.T) {
if err != nil {
t.Fatalf("creating index: %v", err)
}
_, err = m0.API.CreateField(ctx, index, field, pilosa.OptFieldTypeInt(0, 100))
_, err = m0.API.CreateField(ctx, index, field, pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
if err != nil {
t.Fatalf("creating field: %v", err)
}

8
docs/data-model.md
View file

@ -165,6 +165,14 @@ Set(3, B=6)
Check out this [blog post](/blog/range-encoded-bitmaps/) for some more details about BSI in Pilosa.
###### BSI Deprecated Format
The original implementation of BSI required a fixed bit depth when creating fields because the existence bit was written to the bit above the highest bit. The second version of BSI moves the existence bit to the beginning, adds a negative bit as the second bit, and shifts all remaining bits up by two.
Pilosa automatically converts all old data to the new format on startup, however, this can cause issues when upgrading Pilosa and then reverting back to an old version. This documentation section exists as a record for anyone who experiences unusual behavior in BSI between versions.
#### Time
Time fields are similar to `set` fields, but in addition to row and column information, they also store a per-bit time value down to a defined granularity. The following example creates a `time` field called "event" which stores timestamp information down to a day granularity.

4
encoding/proto/proto.go
View file

@ -532,6 +532,8 @@ func encodeFieldOptions(o *pilosa.FieldOptions) *internal.FieldOptions {
CacheSize: o.CacheSize,
Min: o.Min,
Max: o.Max,
Base: o.Base,
BitDepth: uint64(o.BitDepth),
TimeQuantum: string(o.TimeQuantum),
Keys: o.Keys,
}
@ -800,6 +802,8 @@ func decodeFieldOptions(options *internal.FieldOptions, m *pilosa.FieldOptions)
m.CacheSize = options.CacheSize
m.Min = options.Min
m.Max = options.Max
m.Base = options.Base
m.BitDepth = uint(options.BitDepth)
m.TimeQuantum = pilosa.TimeQuantum(options.TimeQuantum)
m.Keys = options.Keys
}

25
executor.go
View file

@ -597,12 +597,12 @@ func (e *executor) executeSumCountShard(ctx context.Context, index string, c *pq
return ValCount{}, nil
}
vsum, vcount, err := fragment.sum(filter, bsig.BitDepth())
vsum, vcount, err := fragment.sum(filter, bsig.BitDepth)
if err != nil {
return ValCount{}, errors.Wrap(err, "computing sum")
}
return ValCount{
Val: int64(vsum) + (int64(vcount) * bsig.Min),
Val: int64(vsum) + (int64(vcount) * bsig.Base),
Count: int64(vcount),
}, nil
}
@ -638,12 +638,12 @@ func (e *executor) executeMinShard(ctx context.Context, index string, c *pql.Cal
return ValCount{}, nil
}
fmin, fcount, err := fragment.min(filter, bsig.BitDepth())
fmin, fcount, err := fragment.min(filter, bsig.BitDepth)
if err != nil {
return ValCount{}, err
}
return ValCount{
Val: int64(fmin) + bsig.Min,
Val: int64(fmin) + bsig.Base,
Count: int64(fcount),
}, nil
}
@ -679,12 +679,12 @@ func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Cal
return ValCount{}, nil
}
fmax, fcount, err := fragment.max(filter, bsig.BitDepth())
fmax, fcount, err := fragment.max(filter, bsig.BitDepth)
if err != nil {
return ValCount{}, err
}
return ValCount{
Val: int64(fmax) + bsig.Min,
Val: int64(fmax) + bsig.Base,
Count: int64(fcount),
}, nil
}
@ -1405,10 +1405,9 @@ func (e *executor) executeRowBSIGroupShard(ctx context.Context, index string, c
return NewRow(), nil
}
return frag.notNull(bsig.BitDepth())
return frag.notNull()
} else if cond.Op == pql.BETWEEN {
predicates, err := cond.IntSliceValue()
if err != nil {
return nil, errors.Wrap(err, "getting condition value")
@ -1443,10 +1442,10 @@ func (e *executor) executeRowBSIGroupShard(ctx context.Context, index string, c
// If the query is asking for the entire valid range, just return
// the not-null bitmap for the bsiGroup.
if predicates[0] <= bsig.Min && predicates[1] >= bsig.Max {
return frag.notNull(bsig.BitDepth())
return frag.notNull()
}
return frag.rangeBetween(bsig.BitDepth(), baseValueMin, baseValueMax)
return frag.rangeBetween(bsig.BitDepth, baseValueMin, baseValueMax)
} else {
@ -1476,16 +1475,16 @@ func (e *executor) executeRowBSIGroupShard(ctx context.Context, index string, c
// LT[E] and GT[E] should return all not-null if selected range fully encompasses valid bsiGroup range.
if (cond.Op == pql.LT && value > bsig.Max) || (cond.Op == pql.LTE && value >= bsig.Max) ||
(cond.Op == pql.GT && value < bsig.Min) || (cond.Op == pql.GTE && value <= bsig.Min) {
return frag.notNull(bsig.BitDepth())
return frag.notNull()
}
// outOfRange for NEQ should return all not-null.
if outOfRange && cond.Op == pql.NEQ {
return frag.notNull(bsig.BitDepth())
return frag.notNull()
}
f.Stats.Count("range:bsigroup", 1, 1.0)
return frag.rangeOp(cond.Op, bsig.BitDepth(), baseValue)
return frag.rangeOp(cond.Op, bsig.BitDepth, baseValue)
}
}

91
executor_test.go
View file

@ -20,6 +20,7 @@ import (
"flag"
"fmt"
"io/ioutil"
"math"
"math/rand"
"reflect"
"strconv"
@ -769,7 +770,7 @@ func TestExecutor_Execute_SetValue(t *testing.T) {
// Create fields.
index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{})
if _, err := index.CreateFieldIfNotExists("f", pilosa.OptFieldTypeInt(0, 50)); err != nil {
if _, err := index.CreateFieldIfNotExists("f", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64)); err != nil {
t.Fatal(err)
} else if _, err := index.CreateFieldIfNotExists("xxx", pilosa.OptFieldTypeDefault()); err != nil {
t.Fatal(err)
@ -806,7 +807,7 @@ func TestExecutor_Execute_SetValue(t *testing.T) {
hldr := test.Holder{Holder: c[0].Server.Holder()}
index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{})
if _, err := index.CreateFieldIfNotExists("f", pilosa.OptFieldTypeInt(0, 100)); err != nil {
if _, err := index.CreateFieldIfNotExists("f", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64)); err != nil {
t.Fatal(err)
}
@ -1214,7 +1215,7 @@ func TestExecutor_Execute_MinMax(t *testing.T) {
t.Fatal(err)
}
if _, err := idx.CreateField("f", pilosa.OptFieldTypeInt(-10, 100)); err != nil {
if _, err := idx.CreateField("f", pilosa.OptFieldTypeInt(-1100, 1000)); err != nil {
t.Fatal(err)
}
@ -1262,32 +1263,6 @@ func TestExecutor_Execute_MinMax(t *testing.T) {
}
}
})
t.Run("Max", func(t *testing.T) {
tests := []struct {
filter string
exp int64
cnt int64
}{
{filter: ``, exp: 60, cnt: 1},
{filter: `Row(x=0)`, exp: 60, cnt: 1},
{filter: `Row(x=1)`, exp: -5, cnt: 1},
{filter: `Row(x=2)`, exp: 40, cnt: 1},
}
for i, tt := range tests {
var pql string
if tt.filter == "" {
pql = `Max(field=f)`
} else {
pql = fmt.Sprintf(`Max(%s, field=f)`, tt.filter)
}
if result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: pql}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(result.Results[0], pilosa.ValCount{Val: tt.exp, Count: tt.cnt}) {
t.Fatalf("unexpected result, test %d: %s", i, spew.Sdump(result))
}
}
})
})
t.Run("ColumnKey", func(t *testing.T) {
@ -1304,7 +1279,7 @@ func TestExecutor_Execute_MinMax(t *testing.T) {
t.Fatal(err)
}
if _, err := idx.CreateField("f", pilosa.OptFieldTypeInt(-10, 100)); err != nil {
if _, err := idx.CreateField("f", pilosa.OptFieldTypeInt(-1110, 1000)); err != nil {
t.Fatal(err)
}
@ -1397,15 +1372,15 @@ func TestExecutor_Execute_Sum(t *testing.T) {
t.Fatal(err)
}
if _, err := idx.CreateField("foo", pilosa.OptFieldTypeInt(10, 100)); err != nil {
if _, err := idx.CreateField("foo", pilosa.OptFieldTypeInt(-990, 1000)); err != nil {
t.Fatal(err)
}
if _, err := idx.CreateField("bar", pilosa.OptFieldTypeInt(0, 100000)); err != nil {
if _, err := idx.CreateField("bar", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64)); err != nil {
t.Fatal(err)
}
if _, err := idx.CreateField("other", pilosa.OptFieldTypeInt(0, 1000)); err != nil {
if _, err := idx.CreateField("other", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64)); err != nil {
t.Fatal(err)
}
@ -1455,15 +1430,15 @@ func TestExecutor_Execute_Sum(t *testing.T) {
t.Fatal(err)
}
if _, err := idx.CreateField("foo", pilosa.OptFieldTypeInt(10, 100)); err != nil {
if _, err := idx.CreateField("foo", pilosa.OptFieldTypeInt(-990, 1000)); err != nil {
t.Fatal(err)
}
if _, err := idx.CreateField("bar", pilosa.OptFieldTypeInt(0, 100000)); err != nil {
if _, err := idx.CreateField("bar", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64)); err != nil {
t.Fatal(err)
}
if _, err := idx.CreateField("other", pilosa.OptFieldTypeInt(0, 1000)); err != nil {
if _, err := idx.CreateField("other", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64)); err != nil {
t.Fatal(err)
}
@ -1857,19 +1832,19 @@ func TestExecutor_Execute_Row_BSIGroup(t *testing.T) {
t.Fatal(err)
}
if _, err := idx.CreateField("foo", pilosa.OptFieldTypeInt(10, 100)); err != nil {
if _, err := idx.CreateField("foo", pilosa.OptFieldTypeInt(-990, 1000)); err != nil {
t.Fatal(err)
}
if _, err := idx.CreateField("bar", pilosa.OptFieldTypeInt(0, 100000)); err != nil {
if _, err := idx.CreateField("bar", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64)); err != nil {
t.Fatal(err)
}
if _, err := idx.CreateField("other", pilosa.OptFieldTypeInt(0, 1000)); err != nil {
if _, err := idx.CreateField("other", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64)); err != nil {
t.Fatal(err)
}
if _, err := idx.CreateField("edge", pilosa.OptFieldTypeInt(-100, 100)); err != nil {
if _, err := idx.CreateField("edge", pilosa.OptFieldTypeInt(-900, 1000)); err != nil {
t.Fatal(err)
}
@ -1893,8 +1868,8 @@ func TestExecutor_Execute_Row_BSIGroup(t *testing.T) {
t.Run("EQ", func(t *testing.T) {
if result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Row(foo == 20)`}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual([]uint64{50, (5 * ShardWidth) + 100}, result.Results[0].(*pilosa.Row).Columns()) {
t.Fatalf("unexpected result: %s", spew.Sdump(result))
} else if got, exp := result.Results[0].(*pilosa.Row).Columns(), []uint64{50, (5 * ShardWidth) + 100}; !reflect.DeepEqual(exp, got) {
t.Fatalf("Query().Row.Columns=%#v, expected %#v", got, exp)
}
})
@ -1903,20 +1878,20 @@ func TestExecutor_Execute_Row_BSIGroup(t *testing.T) {
if result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Row(other != null)`}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual([]uint64{0}, result.Results[0].(*pilosa.Row).Columns()) {
t.Fatalf("unexpected result: %s", spew.Sdump(result))
t.Fatalf("unexpected result: %#v", result.Results[0].(*pilosa.Row).Columns())
}
// NEQ <int>
if result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Row(foo != 20)`}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual([]uint64{ShardWidth, ShardWidth + 1, ShardWidth + 2}, result.Results[0].(*pilosa.Row).Columns()) {
t.Fatalf("unexpected result: %s", spew.Sdump(result))
t.Fatalf("unexpected result: %#v", result.Results[0].(*pilosa.Row).Columns())
}
// NEQ -<int>
if result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Row(other != -20)`}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual([]uint64{0}, result.Results[0].(*pilosa.Row).Columns()) {
//t.Fatalf("unexpected result: %s", spew.Sdump(result))
t.Fatalf("unexpected result: %v", result.Results[0].(*pilosa.Row).Columns())
t.Fatalf("unexpected result: %#v", result.Results[0].(*pilosa.Row).Columns())
}
})
@ -1931,8 +1906,8 @@ func TestExecutor_Execute_Row_BSIGroup(t *testing.T) {
t.Run("LTE", func(t *testing.T) {
if result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Row(foo <= 20)`}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual([]uint64{50, ShardWidth + 2, (5 * ShardWidth) + 100}, result.Results[0].(*pilosa.Row).Columns()) {
t.Fatalf("unexpected result: %s", spew.Sdump(result))
} else if got, exp := result.Results[0].(*pilosa.Row).Columns(), []uint64{50, ShardWidth + 2, (5 * ShardWidth) + 100}; !reflect.DeepEqual(got, exp) {
t.Fatalf("unexpected result: got=%v, exp=%v", got, exp)
}
})
@ -1940,7 +1915,7 @@ func TestExecutor_Execute_Row_BSIGroup(t *testing.T) {
if result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Row(foo > 20)`}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual([]uint64{ShardWidth, ShardWidth + 1}, result.Results[0].(*pilosa.Row).Columns()) {
t.Fatalf("unexpected result: %s", spew.Sdump(result))
t.Fatalf("unexpected result: %v", result.Results[0].(*pilosa.Row).Columns())
}
})
@ -1948,7 +1923,7 @@ func TestExecutor_Execute_Row_BSIGroup(t *testing.T) {
if result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Row(foo >= 20)`}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual([]uint64{50, ShardWidth, ShardWidth + 1, (5 * ShardWidth) + 100}, result.Results[0].(*pilosa.Row).Columns()) {
t.Fatalf("unexpected result: %s", spew.Sdump(result))
t.Fatalf("unexpected result: %v", result.Results[0].(*pilosa.Row).Columns())
}
})
@ -1968,7 +1943,7 @@ func TestExecutor_Execute_Row_BSIGroup(t *testing.T) {
{q: `Row(1000 < other <= 1000)`, exp: false},
{q: `Row(1000 < other < 2000)`, exp: false},
{q: `Row(1000 <= other < 2000)`, exp: true},
{q: `Row(1000 <= other < 20000)`, exp: true},
{q: `Row(1000 <= other <= 2000)`, exp: true},
{q: `Row(1000 < other <= 2000)`, exp: false},
}
@ -1981,7 +1956,7 @@ func TestExecutor_Execute_Row_BSIGroup(t *testing.T) {
if result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: test.q}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(expected, result.Results[0].(*pilosa.Row).Columns()) {
t.Fatalf("unexpected result for query: %s", test.q)
t.Fatalf("unexpected result for query: %s (%#v)", test.q, result.Results[0].(*pilosa.Row).Columns())
}
})
}
@ -2022,7 +1997,7 @@ func TestExecutor_Execute_Row_BSIGroup(t *testing.T) {
})
t.Run("GTBelowMin", func(t *testing.T) {
if result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Row(edge > -200)`}); err != nil {
if result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Row(edge > -1000)`}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual([]uint64{0, 1}, result.Results[0].(*pilosa.Row).Columns()) {
t.Fatalf("unexpected result: %s", spew.Sdump(result.Results[0].(*pilosa.Row).Columns()))
@ -2051,19 +2026,19 @@ func TestExecutor_Execute_Range_BSIGroup_Deprecated(t *testing.T) {
t.Fatal(err)
}
if _, err := idx.CreateField("foo", pilosa.OptFieldTypeInt(10, 100)); err != nil {
if _, err := idx.CreateField("foo", pilosa.OptFieldTypeInt(-990, 1000)); err != nil {
t.Fatal(err)
}
if _, err := idx.CreateField("bar", pilosa.OptFieldTypeInt(0, 100000)); err != nil {
if _, err := idx.CreateField("bar", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64)); err != nil {
t.Fatal(err)
}
if _, err := idx.CreateField("other", pilosa.OptFieldTypeInt(0, 1000)); err != nil {
if _, err := idx.CreateField("other", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64)); err != nil {
t.Fatal(err)
}
if _, err := idx.CreateField("edge", pilosa.OptFieldTypeInt(-100, 100)); err != nil {
if _, err := idx.CreateField("edge", pilosa.OptFieldTypeInt(-1100, 1000)); err != nil {
t.Fatal(err)
}
@ -2188,7 +2163,7 @@ func TestExecutor_Execute_Range_BSIGroup_Deprecated(t *testing.T) {
})
t.Run("GTBelowMin", func(t *testing.T) {
if result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Range(edge > -200)`}); err != nil {
if result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Range(edge > -1200)`}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual([]uint64{0, 1}, result.Results[0].(*pilosa.Row).Columns()) {
t.Fatalf("unexpected result: %s", spew.Sdump(result.Results[0].(*pilosa.Row).Columns()))
@ -2821,7 +2796,7 @@ func TestExecutor_Execute_ClearRow(t *testing.T) {
defer c.Close()
hldr := test.Holder{Holder: c[0].Server.Holder()}
index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{TrackExistence: true})
_, err := index.CreateField("f", pilosa.OptFieldTypeInt(0, 100))
_, err := index.CreateField("f", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
if err != nil {
t.Fatal(err)
}

245
field.go
View file

@ -135,9 +135,6 @@ func OptFieldTypeInt(min, max int64) FieldOption {
if fo.Type != "" {
return errors.Errorf("field type is already set to: %s", fo.Type)
}
if min > max {
return ErrInvalidBSIGroupRange
}
fo.Type = FieldTypeInt
fo.Min = min
fo.Max = max
@ -448,6 +445,22 @@ func (f *Field) openViews() error {
if err := view.open(); err != nil {
return fmt.Errorf("opening view: view=%s, err=%s", view.name, err)
}
// Automatically upgrade BSI v1 fragments if they exist & reopen view.
if bsig := f.bsiGroup(f.name); bsig != nil {
if ok, err := upgradeViewBSIv2(view, bsig.BitDepth); err != nil {
return errors.Wrap(err, "upgrade view bsi v2")
} else if ok {
if err := view.close(); err != nil {
return errors.Wrap(err, "closing upgraded view")
}
view = f.newView(f.viewPath(name), name)
if err := view.open(); err != nil {
return fmt.Errorf("re-opening view: view=%s, err=%s", view.name, err)
}
}
}
view.rowAttrStore = f.rowAttrStore
f.logger.Debugf("add index/field/view to field.viewMap: %s/%s/%s", f.index, f.name, view.name)
f.viewMap[view.name] = view
@ -472,12 +485,23 @@ func (f *Field) loadMeta() error {
}
}
// Initialize "base" to "min" when upgrading from v1 BSI format.
if pb.BitDepth == 0 {
pb.Base = pb.Min
pb.BitDepth = uint64(bitDepthInt64(pb.Max - pb.Min))
if pb.BitDepth == 0 {
pb.BitDepth = 1
}
}
// Copy metadata fields.
f.options.Type = pb.Type
f.options.CacheType = pb.CacheType
f.options.CacheSize = pb.CacheSize
f.options.Min = pb.Min
f.options.Max = pb.Max
f.options.Base = pb.Base
f.options.BitDepth = uint(pb.BitDepth)
f.options.TimeQuantum = TimeQuantum(pb.TimeQuantum)
f.options.Keys = pb.Keys
f.options.NoStandardView = pb.NoStandardView
@ -523,6 +547,8 @@ func (f *Field) applyOptions(opt FieldOptions) error {
}
f.options.Min = 0
f.options.Max = 0
f.options.Base = 0
f.options.BitDepth = 0
f.options.TimeQuantum = ""
f.options.Keys = opt.Keys
case FieldTypeInt:
@ -531,15 +557,19 @@ func (f *Field) applyOptions(opt FieldOptions) error {
f.options.CacheSize = 0
f.options.Min = opt.Min
f.options.Max = opt.Max
f.options.Base = opt.Base
f.options.BitDepth = opt.BitDepth
f.options.TimeQuantum = ""
f.options.Keys = opt.Keys
// Create new bsiGroup.
bsig := &bsiGroup{
Name: f.name,
Type: bsiGroupTypeInt,
Min: opt.Min,
Max: opt.Max,
Name: f.name,
Type: bsiGroupTypeInt,
Min: opt.Min,
Max: opt.Max,
Base: opt.Base,
BitDepth: opt.BitDepth,
}
// Validate bsiGroup.
if err := bsig.validate(); err != nil {
@ -554,6 +584,8 @@ func (f *Field) applyOptions(opt FieldOptions) error {
f.options.CacheSize = 0
f.options.Min = 0
f.options.Max = 0
f.options.Base = 0
f.options.BitDepth = 0
f.options.Keys = opt.Keys
f.options.NoStandardView = opt.NoStandardView
// Set the time quantum.
@ -567,6 +599,8 @@ func (f *Field) applyOptions(opt FieldOptions) error {
f.options.CacheSize = 0
f.options.Min = 0
f.options.Max = 0
f.options.Base = 0
f.options.BitDepth = 0
f.options.TimeQuantum = ""
f.options.Keys = false
default:
@ -965,18 +999,18 @@ func (f *Field) Value(columnID uint64) (value int64, exists bool, err error) {
return 0, false, nil
}
v, exists, err := view.value(columnID, bsig.BitDepth())
v, exists, err := view.value(columnID, bsig.BitDepth)
if err != nil {
return 0, false, err
} else if !exists {
return 0, false, nil
}
return int64(v) + bsig.Min, true, nil
return int64(v) + bsig.Base, true, nil
}
// SetValue sets a field value for a column.
func (f *Field) SetValue(columnID uint64, value int64) (changed bool, err error) {
// Fetch bsiGroup and validate value.
// Fetch bsiGroup & validate min/max.
bsig := f.bsiGroup(f.name)
if bsig == nil {
return false, ErrBSIGroupNotFound
@ -986,16 +1020,37 @@ func (f *Field) SetValue(columnID uint64, value int64) (changed bool, err error)
return false, ErrBSIGroupValueTooHigh
}
// Determine base value to store.
baseValue := int64(value - bsig.Base)
requiredBitDepth := bitDepthInt64(baseValue)
// Increase bit depth value if the unsigned value is greater.
if requiredBitDepth > bsig.BitDepth {
if err := func() error {
f.mu.Lock()
defer f.mu.Unlock()
uvalue := uint64(baseValue)
if value < 0 {
uvalue = uint64(-baseValue)
}
bitDepth := bitDepth(uvalue)
bsig.BitDepth = bitDepth
f.options.BitDepth = bitDepth
return f.saveMeta()
}(); err != nil {
return false, errors.Wrap(err, "increasing bsi max")
}
}
// Fetch target view.
view, err := f.createViewIfNotExists(viewBSIGroupPrefix + f.name)
if err != nil {
return false, errors.Wrap(err, "creating view")
}
// Determine base value to store.
baseValue := uint64(value - bsig.Min)
return view.setValue(columnID, bsig.BitDepth(), baseValue)
return view.setValue(columnID, bsig.BitDepth, baseValue)
}
// Sum returns the sum and count for a field.
@ -1011,11 +1066,11 @@ func (f *Field) Sum(filter *Row, name string) (sum, count int64, err error) {
return 0, 0, nil
}
vsum, vcount, err := view.sum(filter, bsig.BitDepth())
vsum, vcount, err := view.sum(filter, bsig.BitDepth)
if err != nil {
return 0, 0, err
}
return int64(vsum) + (int64(vcount) * bsig.Min), int64(vcount), nil
return int64(vsum) + (int64(vcount) * bsig.Base), int64(vcount), nil
}
// Min returns the min for a field.
@ -1031,11 +1086,11 @@ func (f *Field) Min(filter *Row, name string) (min, count int64, err error) {
return 0, 0, nil
}
vmin, vcount, err := view.min(filter, bsig.BitDepth())
vmin, vcount, err := view.min(filter, bsig.BitDepth)
if err != nil {
return 0, 0, err
}
return int64(vmin) + bsig.Min, int64(vcount), nil
return int64(vmin) + bsig.Base, int64(vcount), nil
}
// Max returns the max for a field.
@ -1051,11 +1106,11 @@ func (f *Field) Max(filter *Row, name string) (max, count int64, err error) {
return 0, 0, nil
}
vmax, vcount, err := view.max(filter, bsig.BitDepth())
vmax, vcount, err := view.max(filter, bsig.BitDepth)
if err != nil {
return 0, 0, err
}
return int64(vmax) + bsig.Min, int64(vcount), nil
return int64(vmax) + bsig.Base, int64(vcount), nil
}
// Range performs a conditional operation on Field.
@ -1079,7 +1134,7 @@ func (f *Field) Range(name string, op pql.Token, predicate int64) (*Row, error)
return NewRow(), nil
}
return view.rangeOp(op, bsig.BitDepth(), baseValue)
return view.rangeOp(op, bsig.BitDepth, baseValue)
}
// Import bulk imports data.
@ -1172,6 +1227,36 @@ func (f *Field) importValue(columnIDs []uint64, values []int64, options *ImportO
return errors.Wrap(ErrBSIGroupNotFound, f.name)
}
// Find the lowest/highest values.
var min, max int64
for i, value := range values {
if i == 0 || value < min {
min = value
}
if i == 0 || value > max {
max = value
}
}
// Determine the highest bit depth required by the min & max.
requiredDepth := bitDepthInt64(min - bsig.Base)
if v := bitDepthInt64(max - bsig.Base); v > requiredDepth {
requiredDepth = v
}
// Increase bit depth if required.
if requiredDepth > bsig.BitDepth {
if err := func() error {
f.mu.Lock()
defer f.mu.Unlock()
bsig.BitDepth = requiredDepth
f.options.BitDepth = requiredDepth
return f.saveMeta()
}(); err != nil {
return errors.Wrap(err, "increasing bsi bit depth")
}
}
// Split import data by fragment.
dataByFragment := make(map[importKey]importValueData)
for i := range columnIDs {
@ -1194,7 +1279,6 @@ func (f *Field) importValue(columnIDs []uint64, values []int64, options *ImportO
// Import into each fragment.
for key, data := range dataByFragment {
// The view must already exist (i.e. we can't create it)
// because we need to know bitDepth (based on min/max value).
view, err := f.createViewIfNotExists(key.View)
@ -1207,12 +1291,12 @@ func (f *Field) importValue(columnIDs []uint64, values []int64, options *ImportO
return errors.Wrap(err, "creating fragment")
}
baseValues := make([]uint64, len(data.Values))
baseValues := make([]int64, len(data.Values))
for i, value := range data.Values {
baseValues[i] = uint64(value - bsig.Min)
baseValues[i] = value - bsig.Base
}
if err := frag.importValue(data.ColumnIDs, baseValues, bsig.BitDepth(), options.Clear); err != nil {
if err := frag.importValue(data.ColumnIDs, baseValues, requiredDepth, options.Clear); err != nil {
return err
}
}
@ -1266,6 +1350,8 @@ func (p fieldInfoSlice) Less(i, j int) bool { return p[i].Name < p[j].Name }
// FieldOptions represents options to set when initializing a field.
type FieldOptions struct {
Base int64 `json:"base,omitempty"`
BitDepth uint `json:"bitDepth,omitempty"`
Min int64 `json:"min,omitempty"`
Max int64 `json:"max,omitempty"`
Keys bool `json:"keys"`
@ -1302,6 +1388,8 @@ func encodeFieldOptions(o *FieldOptions) *internal.FieldOptions {
Type: o.Type,
CacheType: o.CacheType,
CacheSize: o.CacheSize,
Base: o.Base,
BitDepth: uint64(o.BitDepth),
Min: o.Min,
Max: o.Max,
TimeQuantum: string(o.TimeQuantum),
@ -1329,12 +1417,16 @@ func (o *FieldOptions) MarshalJSON() ([]byte, error) {
})
case FieldTypeInt:
return json.Marshal(struct {
Type string `json:"type"`
Min int64 `json:"min"`
Max int64 `json:"max"`
Keys bool `json:"keys"`
Type string `json:"type"`
Base int64 `json:"base"`
BitDepth uint `json:"bitDepth"`
Min int64 `json:"min"`
Max int64 `json:"max"`
Keys bool `json:"keys"`
}{
o.Type,
o.Base,
o.BitDepth,
o.Min,
o.Max,
o.Keys,
@ -1389,20 +1481,12 @@ func isValidBSIGroupType(v string) bool {
// bsiGroup represents a group of range-encoded rows on a field.
type bsiGroup struct {
Name string `json:"name,omitempty"`
Type string `json:"type,omitempty"`
Min int64 `json:"min,omitempty"`
Max int64 `json:"max,omitempty"`
}
// BitDepth returns the number of bits required to store a value between min & max.
func (b *bsiGroup) BitDepth() uint {
for i := uint(0); i < 63; i++ {
if b.Max-b.Min < (1 << i) {
return i
}
}
return 63
Name string `json:"name,omitempty"`
Type string `json:"type,omitempty"`
Min int64 `json:"min,omitempty"`
Max int64 `json:"max,omitempty"`
Base int64 `json:"base,omitempty"`
BitDepth uint `json:"bitDepth,omitempty"`
}
// baseValue adjusts the value to align with the range for Field for a certain
@ -1417,46 +1501,47 @@ func (b *bsiGroup) BitDepth() uint {
// In order to make this work, we effectively need to change the operator to LTE.
// Executor.executeBSIGroupRangeShard() takes this into account and returns
// `frag.FieldNotNull(bsig.BitDepth())` in such instances.
func (b *bsiGroup) baseValue(op pql.Token, value int64) (baseValue uint64, outOfRange bool) {
func (b *bsiGroup) baseValue(op pql.Token, value int64) (baseValue int64, outOfRange bool) {
min, max := b.bitDepthMin(), b.bitDepthMax()
if op == pql.GT || op == pql.GTE {
if value > b.Max {
if value > max {
return baseValue, true
} else if value > b.Min {
baseValue = uint64(value - b.Min)
} else if value > min {
baseValue = int64(value - b.Base)
}
} else if op == pql.LT || op == pql.LTE {
if value < b.Min {
if value < min {
return baseValue, true
} else if value > b.Max {
baseValue = uint64(b.Max - b.Min)
} else if value > max {
baseValue = int64(max - b.Base)
} else {
baseValue = uint64(value - b.Min)
baseValue = int64(value - b.Base)
}
} else if op == pql.EQ || op == pql.NEQ {
if value < b.Min || value > b.Max {
if value < min || value > max {
return baseValue, true
}
baseValue = uint64(value - b.Min)
baseValue = int64(value - b.Base)
}
return baseValue, false
}
// baseValueBetween adjusts the min/max value to align with the range for Field.
func (b *bsiGroup) baseValueBetween(min, max int64) (baseValueMin, baseValueMax uint64, outOfRange bool) {
if max < b.Min || min > b.Max {
return baseValueMin, baseValueMax, true
func (b *bsiGroup) baseValueBetween(lo, hi int64) (baseValueLo, baseValueHi int64, outOfRange bool) {
min, max := b.bitDepthMin(), b.bitDepthMax()
if hi < min || lo > max {
return 0, 0, true
}
// Adjust min/max to range.
if min > b.Min {
baseValueMin = uint64(min - b.Min)
// Limit lo/hi to possible bit range.
if lo < min {
lo = min
}
// Make sure the high value of the BETWEEN does not exceed BitDepth.
if max > b.Max {
baseValueMax = uint64(b.Max - b.Min)
} else if max > b.Min {
baseValueMax = uint64(max - b.Min)
if hi > max {
hi = max
}
return baseValueMin, baseValueMax, false
return lo - b.Base, hi - b.Base, false
}
func (b *bsiGroup) validate() error {
@ -1464,12 +1549,20 @@ func (b *bsiGroup) validate() error {
return ErrBSIGroupNameRequired
} else if !isValidBSIGroupType(b.Type) {
return ErrInvalidBSIGroupType
} else if b.Min > b.Max {
return ErrInvalidBSIGroupRange
}
return nil
}
// bitDepthMin returns the minimum value possible for the current bit depth.
func (b *bsiGroup) bitDepthMin() int64 {
return b.Base - (1 << b.BitDepth) + 1
}
// bitDepthMax returns the maximum value possible for the current bit depth.
func (b *bsiGroup) bitDepthMax() int64 {
return b.Base + (1 << b.BitDepth) - 1
}
// Cache types.
const (
CacheTypeLRU = "lru"
@ -1486,3 +1579,21 @@ func isValidCacheType(v string) bool {
return false
}
}
// bitDepth returns the number of bits required to store a value.
func bitDepth(v uint64) uint {
for i := uint(0); i < 63; i++ {
if v < (1 << i) {
return i
}
}
return 63
}
// bitDepthInt64 returns the required bit depth for abs(v).
func bitDepthInt64(v int64) uint {
if v < 0 {
return bitDepth(uint64(-v))
}
return bitDepth(uint64(v))
}

122
field_internal_test.go
View file

@ -15,7 +15,9 @@
package pilosa
import (
"fmt"
"io/ioutil"
"math"
"os"
"reflect"
"testing"
@ -28,125 +30,127 @@ import (
// Ensure a bsiGroup can adjust to its baseValue.
func TestBSIGroup_BaseValue(t *testing.T) {
b0 := &bsiGroup{
Name: "b0",
Type: bsiGroupTypeInt,
Min: -100,
Max: 900,
Name: "b0",
Type: bsiGroupTypeInt,
Base: -100,
BitDepth: 10,
Min: -1000,
Max: 1000,
}
b1 := &bsiGroup{
Name: "b1",
Type: bsiGroupTypeInt,
Min: 0,
Max: 1000,
Name: "b1",
Type: bsiGroupTypeInt,
Base: 0,
BitDepth: 8,
Min: -255,
Max: 255,
}
b2 := &bsiGroup{
Name: "b2",
Type: bsiGroupTypeInt,
Min: 100,
Max: 1100,
Name: "b2",
Type: bsiGroupTypeInt,
Base: 100,
BitDepth: 11,
Min: math.MinInt64,
Max: math.MaxInt64,
}
t.Run("Normal Condition", func(t *testing.T) {
for _, tt := range []struct {
for i, tt := range []struct {
f *bsiGroup
op pql.Token
val int64
expBaseValue uint64
expBaseValue int64
expOutOfRange bool
}{
// LT
{b0, pql.LT, 5, 105, false},
{b0, pql.LT, -8, 92, false},
{b0, pql.LT, -108, 0, true},
{b0, pql.LT, 1005, 1000, false},
{b0, pql.LT, -108, -8, false},
{b0, pql.LT, 1005, 1023, false},
{b0, pql.LT, 0, 100, false},
{b1, pql.LT, 5, 5, false},
{b1, pql.LT, -8, 0, true},
{b1, pql.LT, 1005, 1000, false},
{b1, pql.LT, -8, -8, false},
{b1, pql.LT, 1005, 255, false},
{b1, pql.LT, 0, 0, false},
{b2, pql.LT, 5, 0, true},
{b2, pql.LT, -8, 0, true},
{b2, pql.LT, 5, -95, false},
{b2, pql.LT, -8, -108, false},
{b2, pql.LT, 105, 5, false},
{b2, pql.LT, 1105, 1000, false},
{b2, pql.LT, 1105, 1005, false},
// GT
{b0, pql.GT, -105, 0, false},
{b0, pql.GT, -5, 95, false},
{b0, pql.GT, 5, 105, false},
{b0, pql.GT, 905, 0, true},
{b0, pql.GT, 905, 1005, false},
{b0, pql.GT, 0, 100, false},
{b1, pql.GT, 5, 5, false},
{b1, pql.GT, -8, 0, false},
{b1, pql.GT, -8, -8, false},
{b1, pql.GT, 1005, 0, true},
{b1, pql.GT, 0, 0, false},
{b2, pql.GT, 5, 0, false},
{b2, pql.GT, -8, 0, false},
{b2, pql.GT, 5, -95, false},
{b2, pql.GT, -8, -108, false},
{b2, pql.GT, 105, 5, false},
{b2, pql.GT, 1105, 0, true},
{b2, pql.GT, 1105, 1005, false},
// EQ
{b0, pql.EQ, -105, 0, true},
{b0, pql.EQ, -105, -5, false},
{b0, pql.EQ, 5, 105, false},
{b0, pql.EQ, 905, 0, true},
{b0, pql.EQ, 905, 1005, false},
{b0, pql.EQ, 0, 100, false},
{b1, pql.EQ, 5, 5, false},
{b1, pql.EQ, -8, 0, true},
{b1, pql.EQ, -8, -8, false},
{b1, pql.EQ, 1005, 0, true},
{b1, pql.EQ, 0, 0, false},
{b2, pql.EQ, 5, 0, true},
{b2, pql.EQ, -8, 0, true},
{b2, pql.EQ, 5, -95, false},
{b2, pql.EQ, -8, -108, false},
{b2, pql.EQ, 105, 5, false},
{b2, pql.EQ, 1105, 0, true},
{b2, pql.EQ, 1105, 1005, false},
} {
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(fmt.Sprint(i), func(t *testing.T) {
bv, oor := tt.f.baseValue(tt.op, tt.val)
if oor != tt.expOutOfRange || !reflect.DeepEqual(bv, tt.expBaseValue) {
t.Errorf("%s) baseValue(%s, %v)=(%v, %v), expected (%v, %v)", tt.f.Name, tt.op, tt.val, bv, oor, tt.expBaseValue, tt.expOutOfRange)
}
})
}
})
t.Run("Betwween Condition", func(t *testing.T) {
for _, tt := range []struct {
t.Run("Between Condition", func(t *testing.T) {
for i, tt := range []struct {
f *bsiGroup
predMin int64
predMax int64
expBaseValueMin uint64
expBaseValueMax uint64
expBaseValueMin int64
expBaseValueMax int64
expOutOfRange bool
}{
{b0, -205, -105, 0, 0, true},
{b0, -105, 80, 0, 180, false},
{b0, -205, -105, -105, -5, false},
{b0, -105, 80, -5, 180, false},
{b0, 5, 20, 105, 120, false},
{b0, 20, 1005, 120, 1000, false},
{b0, 20, 1005, 120, 1023, false},
{b0, 1005, 2000, 0, 0, true},
{b1, -105, -5, 0, 0, true},
{b1, -5, 20, 0, 20, false},
{b1, -105, -5, -105, -5, false},
{b1, -5, 20, -5, 20, false},
{b1, 5, 20, 5, 20, false},
{b1, 20, 1005, 20, 1000, false},
{b1, 20, 1005, 20, 255, false},
{b1, 1005, 2000, 0, 0, true},
{b2, 5, 95, 0, 0, true},
{b2, 95, 120, 0, 20, false},
{b2, 5, 95, -95, -5, false},
{b2, 95, 120, -5, 20, false},
{b2, 105, 120, 5, 20, false},
{b2, 120, 1105, 20, 1000, false},
{b2, 1105, 2000, 0, 0, true},
{b2, 120, 1105, 20, 1005, false},
{b2, 1105, 2000, 1005, 1900, false},
} {
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)
if !reflect.DeepEqual(min, tt.expBaseValueMin) || !reflect.DeepEqual(max, tt.expBaseValueMax) || oor != tt.expOutOfRange {
t.Errorf("%d. %s) baseValueBetween(%v, %v)=(%v, %v, %v), expected (%v, %v, %v)", i, tt.f.Name, tt.predMin, tt.predMax, min, max, oor, tt.expBaseValueMin, tt.expBaseValueMax, tt.expOutOfRange)
}
}
})

5
field_test.go
View file

@ -16,6 +16,7 @@ package pilosa_test
import (
"io/ioutil"
"math"
"testing"
"github.com/google/go-cmp/cmp"
@ -30,7 +31,7 @@ func TestField_SetValue(t *testing.T) {
idx := test.MustOpenIndex()
defer idx.Close()
f, err := idx.CreateField("f", pilosa.OptFieldTypeInt(0, 30))
f, err := idx.CreateField("f", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
if err != nil {
t.Fatal(err)
}
@ -63,7 +64,7 @@ func TestField_SetValue(t *testing.T) {
idx := test.MustOpenIndex()
defer idx.Close()
f, err := idx.CreateField("f", pilosa.OptFieldTypeInt(0, 30))
f, err := idx.CreateField("f", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
if err != nil {
t.Fatal(err)
}

461
fragment.go
View file

@ -83,6 +83,14 @@ const (
// Row ids used for boolean fields.
falseRowID = uint64(0)
trueRowID = uint64(1)
// BSI bits used to check existence & sign.
bsiExistsBit = 0
bsiSignBit = 1
bsiOffsetBit = 2
// Roaring bitmap flags.
roaringFlagBSIv2 = 0x01 // indicates version using low bit for existence
)
// fragment represents the intersection of a field and shard in an index.
@ -97,6 +105,7 @@ type fragment struct {
// File-backed storage
path string
flags byte // user-defined flags passed to roaring
file *os.File
storage *roaring.Bitmap
storageData []byte
@ -136,13 +145,14 @@ type fragment struct {
}
// newFragment returns a new instance of Fragment.
func newFragment(path, index, field, view string, shard uint64) *fragment {
func newFragment(path, index, field, view string, shard uint64, flags byte) *fragment {
return &fragment{
path: path,
index: index,
field: field,
view: view,
shard: shard,
flags: flags,
CacheType: DefaultCacheType,
CacheSize: DefaultCacheSize,
@ -208,6 +218,7 @@ func (f *fragment) openStorage() error {
// Create a roaring bitmap to serve as storage for the shard.
if f.storage == nil {
f.storage = roaring.NewFileBitmap()
f.storage.Flags = f.flags
}
// Open the data file to be mmap'd and used as an ops log.
file, mustClose, err := syswrap.OpenFile(f.path, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666)
@ -679,12 +690,12 @@ func (f *fragment) bit(rowID, columnID uint64) (bool, error) {
}
// value uses a column of bits to read a multi-bit value.
func (f *fragment) value(columnID uint64, bitDepth uint) (value uint64, exists bool, err error) {
func (f *fragment) value(columnID uint64, bitDepth uint) (value int64, exists bool, err error) {
f.mu.Lock()
defer f.mu.Unlock()
// If existence bit is unset then ignore remaining bits.
if v, err := f.bit(uint64(bitDepth), columnID); err != nil {
if v, err := f.bit(bsiExistsBit, columnID); err != nil {
return 0, false, errors.Wrap(err, "getting existence bit")
} else if !v {
return 0, false, nil
@ -692,55 +703,75 @@ func (f *fragment) value(columnID uint64, bitDepth uint) (value uint64, exists b
// Compute other bits into a value.
for i := uint(0); i < bitDepth; i++ {
if v, err := f.bit(uint64(i), columnID); err != nil {
if v, err := f.bit(uint64(bsiOffsetBit+i), columnID); err != nil {
return 0, false, errors.Wrapf(err, "getting value bit %d", i)
} else if v {
value |= (1 << i)
}
}
// Negate if sign bit set.
if v, err := f.bit(bsiSignBit, columnID); err != nil {
return 0, false, errors.Wrap(err, "getting sign bit")
} else if v {
value = -value
}
return value, true, nil
}
// clearValue uses a column of bits to clear a multi-bit value.
func (f *fragment) clearValue(columnID uint64, bitDepth uint, value uint64) (changed bool, err error) {
func (f *fragment) clearValue(columnID uint64, bitDepth uint, value int64) (changed bool, err error) {
return f.setValueBase(columnID, bitDepth, value, true)
}
// setValue uses a column of bits to set a multi-bit value.
func (f *fragment) setValue(columnID uint64, bitDepth uint, value uint64) (changed bool, err error) {
func (f *fragment) setValue(columnID uint64, bitDepth uint, value int64) (changed bool, err error) {
return f.setValueBase(columnID, bitDepth, value, false)
}
func (f *fragment) positionsForValue(columnID uint64, bitDepth uint, value uint64, clear bool, toSet, toClear []uint64) ([]uint64, []uint64, error) {
func (f *fragment) positionsForValue(columnID uint64, bitDepth uint, value int64, clear bool, toSet, toClear []uint64) ([]uint64, []uint64, error) {
// Convert value to an unsigned representation.
uvalue := uint64(value)
if value < 0 {
uvalue = uint64(-value)
}
// Mark value as set.
if bit, err := f.pos(bsiExistsBit, columnID); err != nil {
return toSet, toClear, errors.Wrap(err, "getting not-null pos")
} else if clear {
toClear = append(toClear, bit)
} else {
toSet = append(toSet, bit)
}
// Mark sign.
if bit, err := f.pos(bsiSignBit, columnID); err != nil {
return toSet, toClear, errors.Wrap(err, "getting sign pos")
} else if value >= 0 || clear {
toClear = append(toClear, bit)
} else {
toSet = append(toSet, bit)
}
for i := uint(0); i < bitDepth; i++ {
bit, err := f.pos(uint64(i), columnID)
bit, err := f.pos(uint64(bsiOffsetBit+i), columnID)
if err != nil {
return toSet, toClear, errors.Wrap(err, "getting pos")
}
if value&(1<<i) != 0 {
if uvalue&(1<<i) != 0 {
toSet = append(toSet, bit)
} else {
toClear = append(toClear, bit)
}
}
// Mark value as set.
bit, err := f.pos(uint64(bitDepth), columnID)
if err != nil {
return toSet, toClear, errors.Wrap(err, "getting not-null pos")
}
if clear {
toClear = append(toClear, bit)
} else {
toSet = append(toSet, bit)
}
return toSet, toClear, nil
}
// TODO get rid of this and use positionsForValue to generate a single write op, and set that with importPositions.
func (f *fragment) setValueBase(columnID uint64, bitDepth uint, value uint64, clear bool) (changed bool, err error) {
func (f *fragment) setValueBase(columnID uint64, bitDepth uint, value int64, clear bool) (changed bool, err error) {
f.mu.Lock()
defer f.mu.Unlock()
mustClose, err := f.reopen()
@ -751,15 +782,21 @@ func (f *fragment) setValueBase(columnID uint64, bitDepth uint, value uint64, cl
defer f.safeClose()
}
// Convert value to an unsigned representation.
uvalue := uint64(value)
if value < 0 {
uvalue = uint64(-value)
}
for i := uint(0); i < bitDepth; i++ {
if value&(1<<i) != 0 {
if c, err := f.unprotectedSetBit(uint64(i), columnID); err != nil {
if uvalue&(1<<i) != 0 {
if c, err := f.unprotectedSetBit(uint64(bsiOffsetBit+i), columnID); err != nil {
return changed, err
} else if c {
changed = true
}
} else {
if c, err := f.unprotectedClearBit(uint64(i), columnID); err != nil {
if c, err := f.unprotectedClearBit(uint64(bsiOffsetBit+i), columnID); err != nil {
return changed, err
} else if c {
changed = true
@ -769,40 +806,58 @@ func (f *fragment) setValueBase(columnID uint64, bitDepth uint, value uint64, cl
// Mark value as set (or cleared).
if clear {
if c, err := f.unprotectedClearBit(uint64(bitDepth), columnID); err != nil {
if c, err := f.unprotectedClearBit(uint64(bsiExistsBit), columnID); err != nil {
return changed, errors.Wrap(err, "clearing not-null")
} else if c {
changed = true
}
} else {
if c, err := f.unprotectedSetBit(uint64(bitDepth), columnID); err != nil {
if c, err := f.unprotectedSetBit(uint64(bsiExistsBit), columnID); err != nil {
return changed, errors.Wrap(err, "marking not-null")
} else if c {
changed = true
}
}
// Mark sign bit (or clear).
if value >= 0 || clear {
if c, err := f.unprotectedClearBit(uint64(bsiSignBit), columnID); err != nil {
return changed, errors.Wrap(err, "clearing sign")
} else if c {
changed = true
}
} else {
if c, err := f.unprotectedSetBit(uint64(bsiSignBit), columnID); err != nil {
return changed, errors.Wrap(err, "marking sign")
} else if c {
changed = true
}
}
return changed, nil
}
// importSetValue is a more efficient SetValue just for imports.
func (f *fragment) importSetValue(columnID uint64, bitDepth uint, value uint64, clear bool) (changed bool, err error) { // nolint: unparam
func (f *fragment) importSetValue(columnID uint64, bitDepth uint, value int64, clear bool) (changed bool, err error) { // nolint: unparam
// Convert value to an unsigned representation.
uvalue := uint64(value)
if value < 0 {
uvalue = uint64(-value)
}
for i := uint(0); i < bitDepth; i++ {
if value&(1<<i) != 0 {
bit, err := f.pos(uint64(i), columnID)
if err != nil {
return changed, errors.Wrap(err, "getting set pos")
}
bit, err := f.pos(uint64(bsiOffsetBit+i), columnID)
if err != nil {
return changed, errors.Wrap(err, "getting pos")
}
if uvalue&(1<<i) != 0 {
if c, err := f.storage.Add(bit); err != nil {
return changed, errors.Wrap(err, "adding")
} else if c {
changed = true
}
} else {
bit, err := f.pos(uint64(i), columnID)
if err != nil {
return changed, errors.Wrap(err, "getting clear pos")
}
if c, err := f.storage.Remove(bit); err != nil {
return changed, errors.Wrap(err, "removing")
} else if c {
@ -812,11 +867,9 @@ func (f *fragment) importSetValue(columnID uint64, bitDepth uint, value uint64,
}
// Mark value as set.
p, err := f.pos(uint64(bitDepth), columnID)
if err != nil {
if p, err := f.pos(uint64(bsiExistsBit), columnID); err != nil {
return changed, errors.Wrap(err, "getting not-null pos")
}
if clear {
} else if clear {
if c, err := f.storage.Remove(p); err != nil {
return changed, errors.Wrap(err, "removing not-null from storage")
} else if c {
@ -830,19 +883,40 @@ func (f *fragment) importSetValue(columnID uint64, bitDepth uint, value uint64,
}
}
// Mark sign bit.
if p, err := f.pos(uint64(bsiSignBit), columnID); err != nil {
return changed, errors.Wrap(err, "getting sign pos")
} else if value >= 0 || clear {
if c, err := f.storage.Remove(p); err != nil {
return changed, errors.Wrap(err, "removing sign from storage")
} else if c {
changed = true
}
} else {
if c, err := f.storage.Add(p); err != nil {
return changed, errors.Wrap(err, "adding sign to storage")
} else if c {
changed = true
}
}
return changed, nil
}
// sum returns the sum of a given bsiGroup as well as the number of columns involved.
// A bitmap can be passed in to optionally filter the computed columns.
func (f *fragment) sum(filter *Row, bitDepth uint) (sum, count uint64, err error) {
func (f *fragment) sum(filter *Row, bitDepth uint) (sum int64, count uint64, err error) {
// Compute count based on the existence row.
consider := f.row(uint64(bitDepth))
consider := f.row(bsiExistsBit)
if filter != nil {
consider = consider.Intersect(filter)
}
count = consider.Count()
// Determine positive & negative sets.
nrow := f.row(bsiSignBit)
prow := consider.Difference(nrow)
// Compute the sum based on the bit count of each row multiplied by the
// place value of each row. For example, 10 bits in the 1's place plus
// 4 bits in the 2's place plus 3 bits in the 4's place equals a total
@ -850,11 +924,16 @@ func (f *fragment) sum(filter *Row, bitDepth uint) (sum, count uint64, err error
//
// 10*(2^0) + 4*(2^1) + 3*(2^2) = 30
//
var cnt uint64
// Execute once for positive numbers and once for negative. Subtract the
// negative sum from the positive sum.
for i := uint(0); i < bitDepth; i++ {
row := f.row(uint64(i))
cnt = row.intersectionCount(consider)
sum += (1 << i) * cnt
row := f.row(uint64(bsiOffsetBit + i))
psum := int64((1 << i) * row.intersectionCount(prow))
nsum := int64((1 << i) * row.intersectionCount(nrow))
// Squash to reduce the possibility of overflow.
sum += psum - nsum
}
return sum, count, nil
@ -862,9 +941,8 @@ func (f *fragment) sum(filter *Row, bitDepth uint) (sum, count uint64, err error
// min returns the min of a given bsiGroup as well as the number of columns involved.
// A bitmap can be passed in to optionally filter the computed columns.
func (f *fragment) min(filter *Row, bitDepth uint) (min, count uint64, err error) {
consider := f.row(uint64(bitDepth))
func (f *fragment) min(filter *Row, bitDepth uint) (min int64, count uint64, err error) {
consider := f.row(bsiExistsBit)
if filter != nil {
consider = consider.Intersect(filter)
}
@ -874,58 +952,79 @@ func (f *fragment) min(filter *Row, bitDepth uint) (min, count uint64, err error
return 0, 0, nil
}
for i := bitDepth; i > uint(0); i-- {
ii := i - 1 // allow for uint range: (bitDepth-1) to 0
row := f.row(uint64(ii))
// If we have negative values, we should find the highest unsigned value
// from that set, then negate it, and return it. For example, if values
// (-1, -2) exist, they are stored unsigned (1,2) with a negative sign bit
// set. We take the highest of that set (2) and negate it and return it.
if row := f.row(bsiSignBit).Intersect(consider); row.Any() {
min, count := f.maxUnsigned(row, bitDepth)
return -min, count, nil
}
x := consider.Difference(row)
count = x.Count()
// Otherwise find lowest positive number.
min, count = f.minUnsigned(consider, bitDepth)
return min, count, nil
}
// minUnsigned the lowest value without considering the sign bit. Filter is required.
func (f *fragment) minUnsigned(filter *Row, bitDepth uint) (min int64, count uint64) {
for i := int(bitDepth - 1); i >= 0; i-- {
row := filter.Difference(f.row(uint64(bsiOffsetBit + i)))
count = row.Count()
if count > 0 {
consider = x
filter = row
} else {
min += (1 << ii)
if ii == 0 {
count = consider.Count()
min += (1 << uint(i))
if i == 0 {
count = filter.Count()
}
}
}
return min, count, nil
return min, count
}
// max returns the max of a given bsiGroup as well as the number of columns involved.
// A bitmap can be passed in to optionally filter the computed columns.
func (f *fragment) max(filter *Row, bitDepth uint) (max, count uint64, err error) {
consider := f.row(uint64(bitDepth))
func (f *fragment) max(filter *Row, bitDepth uint) (max int64, count uint64, err error) {
consider := f.row(bsiExistsBit)
if filter != nil {
consider = consider.Intersect(filter)
}
// If there are no columns to consider, return early.
if consider.Count() == 0 {
if !consider.Any() {
return 0, 0, nil
}
for i := bitDepth; i > uint(0); i-- {
ii := i - 1 // allow for uint range: (bitDepth-1) to 0
row := f.row(uint64(ii))
x := row.Intersect(consider)
count = x.Count()
if count > 0 {
max += (1 << ii)
consider = x
} else if ii == 0 {
count = consider.Count()
}
// Find lowest negative number w/o sign and negate, if no positives are available.
pos := consider.Difference(f.row(bsiSignBit))
if !pos.Any() {
max, count = f.minUnsigned(consider, bitDepth)
return -max, count, nil
}
// Otherwise find highest positive number.
max, count = f.maxUnsigned(pos, bitDepth)
return max, count, nil
}
// maxUnsigned the highest value without considering the sign bit. Filter is required.
func (f *fragment) maxUnsigned(filter *Row, bitDepth uint) (max int64, count uint64) {
for i := int(bitDepth - 1); i >= 0; i-- {
row := f.row(uint64(bsiOffsetBit + i)).Intersect(filter)
count = row.Count()
if count > 0 {
max += (1 << uint(i))
filter = row
} else if i == 0 {
count = filter.Count()
}
}
return max, count
}
// rangeOp returns bitmaps with a bsiGroup value encoding matching the predicate.
func (f *fragment) rangeOp(op pql.Token, bitDepth uint, predicate uint64) (*Row, error) {
func (f *fragment) rangeOp(op pql.Token, bitDepth uint, predicate int64) (*Row, error) {
switch op {
case pql.EQ:
return f.rangeEQ(bitDepth, predicate)
@ -940,14 +1039,23 @@ func (f *fragment) rangeOp(op pql.Token, bitDepth uint, predicate uint64) (*Row,
}
}
func (f *fragment) rangeEQ(bitDepth uint, predicate uint64) (*Row, error) {
func (f *fragment) rangeEQ(bitDepth uint, predicate int64) (*Row, error) {
// Start with set of columns with values set.
b := f.row(uint64(bitDepth))
b := f.row(bsiExistsBit)
// Filter to only positive/negative numbers.
upredicate := uint64(predicate)
if predicate < 0 {
upredicate = uint64(-predicate)
b = b.Intersect(f.row(bsiSignBit)) // only negatives
} else {
b = b.Difference(f.row(bsiSignBit)) // only positives
}
// Filter any bits that don't match the current bit value.
for i := int(bitDepth - 1); i >= 0; i-- {
row := f.row(uint64(i))
bit := (predicate >> uint(i)) & 1
row := f.row(uint64(bsiOffsetBit + i))
bit := (upredicate >> uint(i)) & 1
if bit == 1 {
b = b.Intersect(row)
@ -959,9 +1067,9 @@ func (f *fragment) rangeEQ(bitDepth uint, predicate uint64) (*Row, error) {
return b, nil
}
func (f *fragment) rangeNEQ(bitDepth uint, predicate uint64) (*Row, error) {
func (f *fragment) rangeNEQ(bitDepth uint, predicate int64) (*Row, error) {
// Start with set of columns with values set.
b := f.row(uint64(bitDepth))
b := f.row(bsiExistsBit)
// Get the equal bitmap.
eq, err := f.rangeEQ(bitDepth, predicate)
@ -975,22 +1083,44 @@ func (f *fragment) rangeNEQ(bitDepth uint, predicate uint64) (*Row, error) {
return b, nil
}
func (f *fragment) rangeLT(bitDepth uint, predicate uint64, allowEquality bool) (*Row, error) {
keep := NewRow()
func (f *fragment) rangeLT(bitDepth uint, predicate int64, allowEquality bool) (*Row, error) {
// Start with set of columns with values set.
b := f.row(uint64(bitDepth))
b := f.row(bsiExistsBit)
// Create predicate without sign bit.
upredicate := uint64(predicate)
if predicate < 0 {
upredicate = uint64(-predicate)
}
// If predicate is positive, return all positives less than predicate and all negatives.
if (predicate >= 0 && allowEquality) || (predicate >= -1 && !allowEquality) {
pos, err := f.rangeLTUnsigned(b.Difference(f.row(bsiSignBit)), bitDepth, upredicate, allowEquality)
if err != nil {
return nil, err
}
neg := f.row(bsiSignBit)
return neg.Union(pos), nil
}
// Otherwise if predicate is negative, return all negatives greater than upredicate.
return f.rangeGTUnsigned(b.Intersect(f.row(bsiSignBit)), bitDepth, upredicate, allowEquality)
}
// rangeLTUnsigned returns all bits LT/LTE the predicate without considering the sign bit.
func (f *fragment) rangeLTUnsigned(filter *Row, bitDepth uint, predicate uint64, allowEquality bool) (*Row, error) {
keep := NewRow()
// Filter any bits that don't match the current bit value.
leadingZeros := true
for i := int(bitDepth - 1); i >= 0; i-- {
row := f.row(uint64(i))
row := f.row(uint64(bsiOffsetBit + i))
bit := (predicate >> uint(i)) & 1
// Remove any columns with higher bits set.
if leadingZeros {
if bit == 0 {
b = b.Difference(row)
filter = filter.Difference(row)
continue
} else {
leadingZeros = false
@ -1004,32 +1134,54 @@ func (f *fragment) rangeLT(bitDepth uint, predicate uint64, allowEquality bool)
if bit == 0 {
return keep, nil
}
return b.Difference(row.Difference(keep)), nil
return filter.Difference(row.Difference(keep)), nil
}
// If bit is zero then remove all set columns not in excluded bitmap.
if bit == 0 {
b = b.Difference(row.Difference(keep))
filter = filter.Difference(row.Difference(keep))
continue
}
// 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 {
keep = keep.Union(b.Difference(row))
keep = keep.Union(filter.Difference(row))
}
}
return b, nil
return filter, nil
}
func (f *fragment) rangeGT(bitDepth uint, predicate uint64, allowEquality bool) (*Row, error) {
b := f.row(uint64(bitDepth))
func (f *fragment) rangeGT(bitDepth uint, predicate int64, allowEquality bool) (*Row, error) {
b := f.row(bsiExistsBit)
// Create predicate without sign bit.
upredicate := uint64(predicate)
if predicate < 0 {
upredicate = uint64(-predicate)
}
// If predicate is positive, return all positives greater than predicate.
if (predicate >= 0 && allowEquality) || (predicate >= -1 && !allowEquality) {
return f.rangeGTUnsigned(b.Difference(f.row(bsiSignBit)), bitDepth, upredicate, allowEquality)
}
// If predicate is negative, return all negatives less than than upredicate and all positives.
neg, err := f.rangeLTUnsigned(b.Intersect(f.row(bsiSignBit)), bitDepth, upredicate, allowEquality)
if err != nil {
return nil, err
}
pos := b.Difference(f.row(bsiSignBit))
return pos.Union(neg), nil
}
func (f *fragment) rangeGTUnsigned(filter *Row, bitDepth uint, predicate uint64, allowEquality bool) (*Row, error) {
keep := NewRow()
// Filter any bits that don't match the current bit value.
for i := int(bitDepth - 1); i >= 0; i-- {
row := f.row(uint64(i))
row := f.row(uint64(bsiOffsetBit + i))
bit := (predicate >> uint(i)) & 1
// Handle last bit differently.
@ -1039,68 +1191,102 @@ func (f *fragment) rangeGT(bitDepth uint, predicate uint64, allowEquality bool)
if bit == 1 {
return keep, nil
}
return b.Difference(b.Difference(row).Difference(keep)), nil
return filter.Difference(filter.Difference(row).Difference(keep)), nil
}
// If bit is set then remove all unset columns not already kept.
if bit == 1 {
b = b.Difference(b.Difference(row).Difference(keep))
filter = filter.Difference(filter.Difference(row).Difference(keep))
continue
}
// 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 {
keep = keep.Union(b.Intersect(row))
keep = keep.Union(filter.Intersect(row))
}
}
return b, nil
return filter, nil
}
// notNull returns the not-null row (stored at bitDepth).
func (f *fragment) notNull(bitDepth uint) (*Row, error) {
return f.row(uint64(bitDepth)), nil
// notNull returns the exists row.
func (f *fragment) notNull() (*Row, error) {
return f.row(bsiExistsBit), nil
}
// rangeBetween returns bitmaps with a bsiGroup value encoding matching any value between predicateMin and predicateMax.
func (f *fragment) rangeBetween(bitDepth uint, predicateMin, predicateMax uint64) (*Row, error) {
b := f.row(uint64(bitDepth))
func (f *fragment) rangeBetween(bitDepth uint, predicateMin, predicateMax int64) (*Row, error) {
b := f.row(bsiExistsBit)
// Convert predicates to unsigned values.
upredicateMin, upredicateMax := uint64(predicateMin), uint64(predicateMax)
if predicateMin < 0 {
upredicateMin = uint64(-predicateMin)
}
if predicateMax < 0 {
upredicateMax = uint64(-predicateMax)
}
// Handle positive-only values.
if predicateMin >= 0 {
return f.rangeBetweenUnsigned(b.Difference(f.row(bsiSignBit)), bitDepth, upredicateMin, upredicateMax)
}
// Handle negative-only values. Swap unsigned min/max predicates.
if predicateMax < 0 {
return f.rangeBetweenUnsigned(b.Intersect(f.row(bsiSignBit)), bitDepth, upredicateMax, upredicateMin)
}
// If predicate crosses positive/negative boundary then handle separately and union.
pos, err := f.rangeLTUnsigned(b.Difference(f.row(bsiSignBit)), bitDepth, upredicateMax, true)
if err != nil {
return nil, err
}
neg, err := f.rangeLTUnsigned(b.Intersect(f.row(bsiSignBit)), bitDepth, upredicateMin, true)
if err != nil {
return nil, err
}
return pos.Union(neg), nil
}
// rangeBetweenUnsigned returns BSI columns for a range of values. Disregards the sign bit.
func (f *fragment) rangeBetweenUnsigned(filter *Row, bitDepth uint, predicateMin, predicateMax uint64) (*Row, error) {
keep1 := NewRow() // GTE
keep2 := NewRow() // 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))
row := f.row(uint64(bsiOffsetBit + 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))
filter = filter.Difference(filter.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))
keep1 = keep1.Union(filter.Intersect(row))
}
}
// LTE predicateMin
// LTE predicateMax
// If bit is zero then remove all set bits not in excluded bitmap.
if bit2 == 0 {
b = b.Difference(row.Difference(keep2))
filter = filter.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))
keep2 = keep2.Union(filter.Difference(row))
}
}
}
return b, nil
return filter, nil
}
// pos translates the row ID and column ID into a position in the storage bitmap.
@ -1735,7 +1921,7 @@ func (f *fragment) bulkImportMutex(rowIDs, columnIDs []uint64) error {
return errors.Wrap(f.importPositions(toSet, toClear, rowSet), "importing positions")
}
func (f *fragment) importValueSmallWrite(columnIDs, values []uint64, bitDepth uint, clear bool) error {
func (f *fragment) importValueSmallWrite(columnIDs []uint64, values []int64, bitDepth uint, clear bool) error {
// TODO figure out how to avoid re-allocating these each time. Probably
// possible to store them on the fragment with a capacity based on
// MaxOpN. For now, we know that the total number of bits to be
@ -1751,6 +1937,7 @@ func (f *fragment) importValueSmallWrite(columnIDs, values []uint64, bitDepth ui
if _, ok := colSet[columnID]; ok {
continue
}
colSet[columnID] = struct{}{}
toSet, toClear, err = f.positionsForValue(columnID, bitDepth, value, clear, toSet, toClear)
if err != nil {
@ -1772,7 +1959,7 @@ func (f *fragment) importValueSmallWrite(columnIDs, values []uint64, bitDepth ui
}
// importValue bulk imports a set of range-encoded values.
func (f *fragment) importValue(columnIDs, values []uint64, bitDepth uint, clear bool) error {
func (f *fragment) importValue(columnIDs []uint64, values []int64, bitDepth uint, clear bool) error {
f.mu.Lock()
defer f.mu.Unlock()
@ -1783,7 +1970,6 @@ func (f *fragment) importValue(columnIDs, values []uint64, bitDepth uint, clear
if len(columnIDs)*int(bitDepth+1)+f.opN < f.MaxOpN {
return errors.Wrap(f.importValueSmallWrite(columnIDs, values, bitDepth, clear), "import small write")
}
f.storage.OpWriter = nil
@ -2288,6 +2474,51 @@ func (f *fragment) rows(start uint64, filters ...rowFilter) []uint64 {
return rows
}
// upgradeRoaringBSIv2 upgrades a fragment that contains old BSI formatting
// to a new BSI format (v2). The new format moves the "exists" bit to the
// beginning & adds a negative sign bit.
func upgradeRoaringBSIv2(f *fragment, bitDepth uint) (string, error) {
// If flag set, already upgraded. Exit.
if f.storage.Flags&roaringFlagBSIv2 == 1 {
return "", nil
}
other := roaring.NewBitmap()
other.Flags = roaringFlagBSIv2
func() {
f.mu.Lock()
defer f.mu.Unlock()
f.storage.ForEach(func(i uint64) {
rowID, columnID := i/ShardWidth, (f.shard*ShardWidth)+(i%ShardWidth)
if rowID == uint64(bitDepth) {
_, _ = other.Add(pos(bsiExistsBit, columnID)) // move exists bit to beginning
} else {
_, _ = other.Add(pos(rowID+bsiOffsetBit, columnID)) // move other bits up
}
})
}()
// Create temporary file next to existing file.
newPath := f.path + ".tmp"
file, err := os.OpenFile(newPath, os.O_WRONLY|os.O_CREATE, 0666)
if err != nil {
return "", err
}
defer file.Close()
// Write & flush to temporary file.
if _, err := other.WriteTo(file); err != nil {
return "", err
} else if err := file.Sync(); err != nil {
return "", err
} else if err := file.Close(); err != nil {
return "", err
}
return newPath, nil
}
type rowIterator struct {
f *fragment
rowIDs []uint64

79
fragment_internal_test.go
View file

@ -276,7 +276,7 @@ func TestFragment_SetValue(t *testing.T) {
}
// Non-existent value.
if value, exists, err := f.value(100, 11); err != nil {
if value, exists, err := f.value(101, 11); err != nil {
t.Fatal(err)
} else if value != 0 {
t.Fatalf("unexpected value: %d", value)
@ -305,7 +305,7 @@ func TestFragment_SetValue(t *testing.T) {
m[columnID] = int64(value)
if _, err := f.setValue(columnID, bitDepth, value); err != nil {
if _, err := f.setValue(columnID, bitDepth, int64(value)); err != nil {
t.Fatal(err)
}
}
@ -409,7 +409,7 @@ func TestFragment_MinMax(t *testing.T) {
t.Run("Min", func(t *testing.T) {
tests := []struct {
filter *Row
exp uint64
exp int64
cnt uint64
}{
{filter: nil, exp: 0, cnt: 1},
@ -433,7 +433,7 @@ func TestFragment_MinMax(t *testing.T) {
t.Run("Max", func(t *testing.T) {
tests := []struct {
filter *Row
exp uint64
exp int64
cnt uint64
}{
{filter: nil, exp: 2818, cnt: 2},
@ -444,12 +444,15 @@ func TestFragment_MinMax(t *testing.T) {
{filter: NewRow(7000), exp: 0, cnt: 1},
}
for i, test := range tests {
var columns []uint64
if test.filter != nil {
columns = test.filter.Columns()
}
if max, cnt, err := f.max(test.filter, bitDepth); err != nil {
t.Fatal(err)
} else if max != test.exp {
t.Errorf("test %d expected max: %v, but got: %v", i, test.exp, max)
} else if cnt != test.cnt {
t.Errorf("test %d expected cnt: %v, but got: %v", i, test.cnt, cnt)
} else if max != test.exp || cnt != test.cnt {
t.Errorf("%d. max(%v, %v)=(%v, %v), expected (%v, %v)", i, columns, bitDepth, max, cnt, test.exp, test.cnt)
}
}
})
@ -657,7 +660,7 @@ func benchmarkSetValues(b *testing.B, bitDepth uint, f *fragment, cfunc func(uin
for i := 0; i < b.N; i++ {
// We're not checking the error because this is a benchmark.
// That does mean the result could be completely wrong...
_, _ = f.setValue(column, bitDepth, uint64(i))
_, _ = f.setValue(column, bitDepth, int64(i))
column = cfunc(column)
}
}
@ -686,9 +689,9 @@ func benchmarkImportValues(b *testing.B, bitDepth uint, f *fragment, cfunc func(
column := uint64(0)
b.StopTimer()
columns := make([]uint64, b.N)
values := make([]uint64, b.N)
values := make([]int64, b.N)
for i := 0; i < b.N; i++ {
values[i] = uint64(i)
values[i] = int64(i)
columns[i] = column
column = cfunc(column)
}
@ -805,23 +808,23 @@ func BenchmarkFragment_RepeatedSmallImportsRoaring(b *testing.B) {
func BenchmarkFragment_RepeatedSmallValueImports(b *testing.B) {
initialCols := make([]uint64, 0, ShardWidth)
initialVals := make([]uint64, 0, ShardWidth)
initialVals := make([]int64, 0, ShardWidth)
for i := uint64(0); i < ShardWidth; i++ {
// every 29 columns, skip between 0 and 12 columns
if i%29 == 0 {
i += i % 13
}
initialCols = append(initialCols, i)
initialVals = append(initialVals, uint64(rand.Int63n(1<<21)))
initialVals = append(initialVals, int64(rand.Int63n(1<<21)))
}
for _, numUpdates := range []int{100} {
for _, valsPerUpdate := range []int{10, 100} {
updateCols := make([]uint64, numUpdates*valsPerUpdate)
updateVals := make([]uint64, numUpdates*valsPerUpdate)
updateVals := make([]int64, numUpdates*valsPerUpdate)
for i := 0; i < numUpdates*valsPerUpdate; i++ {
updateCols[i] = uint64(rand.Int63n(ShardWidth))
updateVals[i] = uint64(rand.Int63n(1 << 21))
updateVals[i] = int64(rand.Int63n(1 << 21))
}
for _, opN := range []int{1, 5000, 50000} {
@ -1372,7 +1375,7 @@ func BenchmarkFragment_Blocks(b *testing.B) {
}
// Open the fragment specified by the path.
f := newFragment(*FragmentPath, "i", "f", viewStandard, 0)
f := newFragment(*FragmentPath, "i", "f", viewStandard, 0, 0)
if err := f.Open(); err != nil {
b.Fatal(err)
}
@ -1901,7 +1904,7 @@ func BenchmarkFragment_Snapshot(b *testing.B) {
b.ReportAllocs()
// Open the fragment specified by the path.
f := newFragment(*FragmentPath, "i", "f", viewStandard, 0)
f := newFragment(*FragmentPath, "i", "f", viewStandard, 0, 0)
if err := f.Open(); err != nil {
b.Fatal(err)
}
@ -2231,7 +2234,7 @@ func BenchmarkImportIntoLargeFragment(b *testing.B) {
}
origF.Close()
fi.Close()
nf := newFragment(fi.Name(), "i", "f", viewStandard, 0)
nf := newFragment(fi.Name(), "i", "f", viewStandard, 0, 0)
err = nf.Open()
if err != nil {
b.Fatalf("opening fragment: %v", err)
@ -2268,7 +2271,7 @@ func BenchmarkImportRoaringIntoLargeFragment(b *testing.B) {
}
origF.Close()
fi.Close()
nf := newFragment(fi.Name(), "i", "f", viewStandard, 0)
nf := newFragment(fi.Name(), "i", "f", viewStandard, 0, 0)
err = nf.Open()
if err != nil {
b.Fatalf("opening fragment: %v", err)
@ -2463,7 +2466,7 @@ func mustOpenFragment(index, field, view string, shard uint64, cacheType string)
cacheType = DefaultCacheType
}
f := newFragment(file.Name(), index, field, view, shard)
f := newFragment(file.Name(), index, field, view, shard, 0)
f.CacheType = cacheType
f.RowAttrStore = &memAttrStore{
store: make(map[uint64]map[string]interface{}),
@ -2988,7 +2991,7 @@ func TestFragmentPositionsForValue(t *testing.T) {
tests := []struct {
columnID uint64
bitDepth uint
value uint64
value int64
clear bool
toSet []uint64
toClear []uint64
@ -2997,43 +3000,43 @@ func TestFragmentPositionsForValue(t *testing.T) {
columnID: 0,
bitDepth: 1,
value: 0,
toSet: []uint64{ShardWidth},
toClear: []uint64{0},
toSet: []uint64{0}, // exists bit only
toClear: []uint64{ShardWidth, ShardWidth * 2}, // sign bit & 1-position
},
{
columnID: 0,
bitDepth: 3,
value: 0,
toSet: []uint64{ShardWidth * 3},
toClear: []uint64{0, ShardWidth, ShardWidth * 2},
toSet: []uint64{0}, // exists bit only
toClear: []uint64{ShardWidth * 1, ShardWidth * 2, ShardWidth * 3, ShardWidth * 4}, // sign bit, 1, 2, 4
},
{
columnID: 1,
bitDepth: 3,
value: 0,
toSet: []uint64{ShardWidth*3 + 1},
toClear: []uint64{1, ShardWidth + 1, ShardWidth*2 + 1},
toSet: []uint64{1}, // exists bit only
toClear: []uint64{ShardWidth + 1, ShardWidth*2 + 1, ShardWidth*3 + 1, ShardWidth*4 + 1}, // sign bit, 1, 2, 4
},
{
columnID: 0,
bitDepth: 1,
value: 1,
toSet: []uint64{0, ShardWidth},
toClear: []uint64{},
toSet: []uint64{0, ShardWidth * 2}, // exists bit, 1
toClear: []uint64{ShardWidth}, // sign bit only
},
{
columnID: 0,
bitDepth: 4,
value: 10,
toSet: []uint64{ShardWidth, ShardWidth * 3, ShardWidth * 4},
toClear: []uint64{0, ShardWidth * 2},
toSet: []uint64{0, ShardWidth * 3, ShardWidth * 5}, // exists bit, 2, 8
toClear: []uint64{ShardWidth * 1, ShardWidth * 2, ShardWidth * 4}, // sign bit, 1, 4
},
{
columnID: 0,
bitDepth: 5,
value: 10,
toSet: []uint64{ShardWidth, ShardWidth * 3, ShardWidth * 5},
toClear: []uint64{0, ShardWidth * 2, ShardWidth * 4},
toSet: []uint64{0, ShardWidth * 3, ShardWidth * 5}, // exists bit, 2, 8
toClear: []uint64{ShardWidth * 1, ShardWidth * 2, ShardWidth * 4, ShardWidth * 6}, // sign bit, 1, 4, 16
},
}
@ -3138,7 +3141,7 @@ func TestImportClearRestart(t *testing.T) {
check(t, f, exp)
f2 := newFragment(f.path, "i", "f", viewStandard, 0)
f2 := newFragment(f.path, "i", "f", viewStandard, 0, 0)
f2.MaxOpN = maxOpN
f2.CacheType = f.CacheType
@ -3172,7 +3175,7 @@ func TestImportClearRestart(t *testing.T) {
check(t, f2, exp)
f3 := newFragment(f2.path, "i", "f", viewStandard, 0)
f3 := newFragment(f2.path, "i", "f", viewStandard, 0, 0)
f3.MaxOpN = maxOpN
f3.CacheType = f.CacheType
@ -3229,7 +3232,7 @@ func TestImportValueConcurrent(t *testing.T) {
i := i
eg.Go(func() error {
for j := uint64(0); j < 10; j++ {
err := f.importValue([]uint64{j}, []uint64{uint64(rand.Int63n(1000))}, 10, i%2 == 0)
err := f.importValue([]uint64{j}, []int64{int64(rand.Int63n(1000))}, 10, i%2 == 0)
if err != nil {
return err
}
@ -3246,14 +3249,14 @@ func TestImportValueConcurrent(t *testing.T) {
func TestImportMultipleValues(t *testing.T) {
tests := []struct {
cols []uint64
vals []uint64
vals []int64
checkCols []uint64
checkVals []uint64
depth uint
}{
{
cols: []uint64{0, 0},
vals: []uint64{97, 100},
vals: []int64{97, 100},
depth: 7,
checkCols: []uint64{0},
checkVals: []uint64{100},

8
http/client_test.go
View file

@ -832,8 +832,8 @@ func TestClient_ImportValue(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if min != -100 || cnt != 0 {
t.Fatalf("unexpected values: got min=%v, count=%v; expected min=-100, cnt=0", min, cnt)
if min != 0 || cnt != 0 {
t.Fatalf("unexpected values: got min=%v, count=%v; expected min=0, cnt=0", min, cnt)
}
// Verify Max.
@ -871,8 +871,8 @@ func TestClient_ImportValue(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if min != -100 || cnt != 0 {
t.Fatalf("unexpected values: got min=%v, count=%v; expected min=-100, cnt=0", min, cnt)
if min != 0 || cnt != 0 {
t.Fatalf("unexpected values: got min=%v, count=%v; expected min=0, cnt=0", min, cnt)
}
// Verify Max.

7
http/handler.go
View file

@ -22,6 +22,7 @@ import (
"fmt"
"io"
"io/ioutil"
"math"
"net"
"net/http"
_ "net/http/pprof" // Imported for its side-effect of registering pprof endpoints with the server.
@ -758,7 +759,7 @@ func (h *Handler) handlePostField(w http.ResponseWriter, r *http.Request) {
case pilosa.FieldTypeSet:
fos = append(fos, pilosa.OptFieldTypeSet(*req.Options.CacheType, *req.Options.CacheSize))
case pilosa.FieldTypeInt:
fos = append(fos, pilosa.OptFieldTypeInt(*req.Options.Min, *req.Options.Max))
fos = append(fos, pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
case pilosa.FieldTypeTime:
fos = append(fos, pilosa.OptFieldTypeTime(*req.Options.TimeQuantum, req.Options.NoStandardView))
case pilosa.FieldTypeMutex:
@ -824,10 +825,6 @@ func (o *fieldOptions) validate() error {
return pilosa.NewBadRequestError(errors.New("cacheType does not apply to field type int"))
} else if o.CacheSize != nil {
return pilosa.NewBadRequestError(errors.New("cacheSize does not apply to field type int"))
} else if o.Min == nil {
return pilosa.NewBadRequestError(errors.New("min is required for field type int"))
} else if o.Max == nil {
return pilosa.NewBadRequestError(errors.New("max is required for field type int"))
} else if o.TimeQuantum != nil {
return pilosa.NewBadRequestError(errors.New("timeQuantum does not apply to field type int"))
}

2
index_test.go
View file

@ -93,7 +93,7 @@ func TestIndex_CreateField(t *testing.T) {
defer index.Close()
// Create field with schema and verify it exists.
if f, err := index.CreateField("f", pilosa.OptFieldTypeInt(10, 20)); err != nil {
if f, err := index.CreateField("f", pilosa.OptFieldTypeInt(-990, 1000)); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(f.Type(), pilosa.FieldTypeInt) {
t.Fatalf("unexpected type: %#v", f.Type())

2161
internal/private.pb.go
View file

File diff suppressed because it is too large Load diff

152
internal/private.proto
View file

@ -8,14 +8,16 @@ message IndexMeta {
}
message FieldOptions {
string Type = 8;
string Type = 8;
string CacheType = 3;
uint32 CacheSize = 4;
int64 Min = 9;
int64 Max = 10;
string TimeQuantum = 5;
bool Keys = 11;
bool NoStandardView = 12;
int64 Min = 9;
int64 Max = 10;
bool Keys = 11;
bool NoStandardView = 12;
int64 Base = 13;
uint64 BitDepth = 14;
}
message ImportResponse {
@ -40,154 +42,154 @@ message Cache {
}
message MaxShards {
map<string, uint64> Standard = 1;
map<string, uint64> Standard = 1;
}
message CreateShardMessage {
string Index = 1;
string Field = 3;
uint64 Shard = 2;
string Index = 1;
string Field = 3;
uint64 Shard = 2;
}
message DeleteIndexMessage {
string Index = 1;
string Index = 1;
}
message CreateIndexMessage {
string Index = 1;
IndexMeta Meta = 2;
string Index = 1;
IndexMeta Meta = 2;
}
message CreateFieldMessage {
string Index = 1;
string Field = 2;
FieldOptions Meta = 3;
string Index = 1;
string Field = 2;
FieldOptions Meta = 3;
}
message DeleteFieldMessage {
string Index = 1;
string Field = 2;
string Index = 1;
string Field = 2;
}
message DeleteAvailableShardMessage {
string Index = 1;
string Field = 2;
uint64 ShardID = 3;
string Index = 1;
string Field = 2;
uint64 ShardID = 3;
}
message Field {
string Name = 1;
FieldOptions Meta = 2;
repeated string Views = 3;
string Name = 1;
FieldOptions Meta = 2;
repeated string Views = 3;
}
message Schema {
repeated Index Indexes = 1;
repeated Index Indexes = 1;
}
message Index {
string Name = 1;
repeated Field Fields = 4;
string Name = 1;
repeated Field Fields = 4;
}
message URI {
string Scheme = 1;
string Host = 2;
uint32 Port = 3;
string Scheme = 1;
string Host = 2;
uint32 Port = 3;
}
message Node {
string ID = 1;
URI URI = 2;
bool IsCoordinator = 3;
string State = 4;
string ID = 1;
URI URI = 2;
bool IsCoordinator = 3;
string State = 4;
}
message NodeStateMessage {
string NodeID = 1;
string State = 2;
string NodeID = 1;
string State = 2;
}
message NodeEventMessage {
uint32 Event = 1;
Node Node = 2;
uint32 Event = 1;
Node Node = 2;
}
message NodeStatus {
Node Node = 1;
Schema Schema = 3;
repeated IndexStatus Indexes = 4;
Node Node = 1;
Schema Schema = 3;
repeated IndexStatus Indexes = 4;
}
message IndexStatus {
string Name = 1;
repeated FieldStatus Fields = 2;
string Name = 1;
repeated FieldStatus Fields = 2;
}
message FieldStatus {
string Name = 1;
repeated uint64 AvailableShards = 2;
string Name = 1;
repeated uint64 AvailableShards = 2;
}
message ClusterStatus {
string ClusterID = 1;
string State = 2;
repeated Node Nodes = 3;
string ClusterID = 1;
string State = 2;
repeated Node Nodes = 3;
}
message BSIGroup {
string Name = 1;
string Type = 2;
int64 Min = 3;
int64 Max = 4;
string Name = 1;
string Type = 2;
int64 Min = 3;
int64 Max = 4;
}
message CreateViewMessage {
string Index = 1;
string Field = 2;
string View = 3;
string Index = 1;
string Field = 2;
string View = 3;
}
message DeleteViewMessage {
string Index = 1;
string Field = 2;
string View = 3;
string Index = 1;
string Field = 2;
string View = 3;
}
message ResizeInstruction {
int64 JobID = 1;
Node Node = 2;
Node Coordinator = 3;
repeated ResizeSource Sources = 4;
NodeStatus NodeStatus = 7;
ClusterStatus ClusterStatus = 6;
int64 JobID = 1;
Node Node = 2;
Node Coordinator = 3;
repeated ResizeSource Sources = 4;
NodeStatus NodeStatus = 7;
ClusterStatus ClusterStatus = 6;
}
message ResizeSource {
Node Node = 1;
string Index = 2;
string Field = 3;
string View = 4;
uint64 Shard = 5;
Node Node = 1;
string Index = 2;
string Field = 3;
string View = 4;
uint64 Shard = 5;
}
message ResizeInstructionComplete {
int64 JobID = 1;
Node Node = 2;
string Error = 3;
int64 JobID = 1;
Node Node = 2;
string Error = 3;
}
message SetCoordinatorMessage {
Node New = 1;
Node New = 1;
}
message UpdateCoordinatorMessage {
Node New = 1;
Node New = 1;
}
message Topology {
string ClusterID = 1;
repeated string NodeIDs = 2;
string ClusterID = 1;
repeated string NodeIDs = 2;
}
message RecalculateCaches {}

1158
internal/public.pb.go
View file

File diff suppressed because it is too large Load diff

44
roaring/roaring.go
View file

@ -31,16 +31,18 @@ const (
// MagicNumber is an identifier, in bytes 0-1 of the file.
MagicNumber = uint32(12348)
// storageVersion indicates the storage version, in bytes 2-3.
// storageVersion indicates the storage version, in byte 2.
storageVersion = uint32(0)
// cookie is the first four bytes in a roaring bitmap file,
// NOTE: byte 3 stores user-defined flags.
// cookie is the first 3 bytes in a roaring bitmap file,
// formed by joining MagicNumber and storageVersion
cookie = MagicNumber + storageVersion<<16
// headerBaseSize is the size in bytes of the cookie and key count at the
// beginning of a file.
headerBaseSize = 4 + 4
// headerBaseSize is the size in bytes of the cookie, flags, and key count
// at the beginning of a file.
headerBaseSize = 3 + 1 + 4
// runCountHeaderSize is the size in bytes of the run count stored
// at the beginning of every serialized run container.
@ -123,6 +125,9 @@ type ContainerIterator interface {
type Bitmap struct {
Containers Containers
// User-defined flags.
Flags byte
// Number of bit change operations written to the writer. Some operations
// contain multiple values, each of those counts the number of values rather
// than counting as one operation.
@ -970,7 +975,7 @@ func (b *Bitmap) writeToUnoptimized(w io.Writer) (n int64, err error) {
n: 0,
}
ew.WriteUint32(byte4, cookie)
ew.WriteUint32(byte4, cookie|(uint32(b.Flags)<<24))
ew.WriteUint32(byte4, uint32(containerCount))
// Descriptive header section: encode keys and cardinality.
@ -1032,7 +1037,8 @@ func (b *Bitmap) unmarshalPilosaRoaring(data []byte) error {
// Verify the first two bytes are a valid MagicNumber, and second two bytes match current storageVersion.
fileMagic := uint32(binary.LittleEndian.Uint16(data[0:2]))
fileVersion := uint32(binary.LittleEndian.Uint16(data[2:4]))
b.Flags = data[2]
fileVersion := uint32(data[3])
if fileMagic != MagicNumber {
return fmt.Errorf("invalid roaring file, magic number %v is incorrect", fileMagic)
}
@ -1041,8 +1047,8 @@ func (b *Bitmap) unmarshalPilosaRoaring(data []byte) error {
return fmt.Errorf("wrong roaring version, file is v%d, server requires v%d", fileVersion, storageVersion)
}
// Read key count in bytes sizeof(cookie):(sizeof(cookie)+sizeof(uint32)).
keyN := binary.LittleEndian.Uint32(data[4:8])
// Read key count in bytes sizeof(cookie)+sizeof(flag):(sizeof(cookie)+sizeof(uint32)).
keyN := binary.LittleEndian.Uint32(data[3+1 : 8])
headerSize := headerBaseSize
b.Containers.Reset()
@ -4166,11 +4172,11 @@ const (
serialCookie = 12347 // runs, arrays, and bitmaps
)
func readOfficialHeader(buf []byte) (size uint32, containerTyper func(index uint, card int) byte, header, pos int, haveRuns bool, err error) {
func readOfficialHeader(buf []byte) (size uint32, containerTyper func(index uint, card int) byte, header, pos int, flags byte, haveRuns bool, err error) {
statsHit("readOfficialHeader")
if len(buf) < 8 {
err = fmt.Errorf("buffer too small, expecting at least 8 bytes, was %d", len(buf))
return size, containerTyper, header, pos, haveRuns, err
return size, containerTyper, header, pos, flags, haveRuns, err
}
cf := func(index uint, card int) (newType byte) {
newType = containerBitmap
@ -4180,7 +4186,8 @@ func readOfficialHeader(buf []byte) (size uint32, containerTyper func(index uint
return newType
}
containerTyper = cf
cookie := binary.LittleEndian.Uint32(buf)
cookie := binary.LittleEndian.Uint32(buf) & 0xFFFFFF
flags = buf[3]
pos += 4
// cookie header
@ -4195,7 +4202,7 @@ func readOfficialHeader(buf []byte) (size uint32, containerTyper func(index uint
isRunBitmapSize := (int(size) + 7) / 8
if pos+isRunBitmapSize > len(buf) {
err = fmt.Errorf("malformed bitmap, is-run bitmap overruns buffer at %d", pos+isRunBitmapSize)
return size, containerTyper, header, pos, haveRuns, err
return size, containerTyper, header, pos, flags, haveRuns, err
}
isRunBitmap := buf[pos : pos+isRunBitmapSize]
@ -4208,22 +4215,22 @@ func readOfficialHeader(buf []byte) (size uint32, containerTyper func(index uint
}
} else {
err = fmt.Errorf("did not find expected serialCookie in header")
return size, containerTyper, header, pos, haveRuns, err
return size, containerTyper, header, pos, flags, haveRuns, err
}
header = pos
if size > (1 << 16) {
err = fmt.Errorf("it is logically impossible to have more than (1<<16) containers")
return size, containerTyper, header, pos, haveRuns, err
return size, containerTyper, header, pos, flags, haveRuns, err
}
// descriptive header
if pos+2*2*int(size) > len(buf) {
err = fmt.Errorf("malformed bitmap, key-cardinality slice overruns buffer at %d", pos+2*2*int(size))
return size, containerTyper, header, pos, haveRuns, err
return size, containerTyper, header, pos, flags, haveRuns, err
}
pos += 2 * 2 * int(size) // moving pos past keycount
return size, containerTyper, header, pos, haveRuns, err
return size, containerTyper, header, pos, flags, haveRuns, err
}
// UnmarshalBinary decodes b from a binary-encoded byte slice. data can be in
@ -4240,10 +4247,11 @@ func (b *Bitmap) UnmarshalBinary(data []byte) error {
return errors.Wrap(b.unmarshalPilosaRoaring(data), "unmarshaling as pilosa roaring")
}
keyN, containerTyper, header, pos, haveRuns, err := readOfficialHeader(data)
keyN, containerTyper, header, pos, flags, haveRuns, err := readOfficialHeader(data)
if err != nil {
return errors.Wrap(err, "reading roaring header")
}
b.Flags = flags
b.Containers.Reset()
// Descriptive header section: Read container keys and cardinalities.

10
row.go
View file

@ -113,6 +113,16 @@ func (r *Row) Intersect(other *Row) *Row {
return &Row{segments: segments}
}
// Any returns true if row contains any bits.
func (r *Row) Any() bool {
for _, s := range r.segments {
if s.data.Any() {
return true
}
}
return false
}
// Xor returns the xor of r and other.
func (r *Row) Xor(other *Row) *Row {
var segments []rowSegment

46
view.go
View file

@ -166,6 +166,15 @@ func (v *view) close() error {
return nil
}
// flags returns a set of flags for the underlying fragments.
func (v *view) flags() byte {
var flag byte
if v.fieldType == FieldTypeInt {
flag |= roaringFlagBSIv2
}
return flag
}
// availableShards returns a bitmap of shards which contain data.
func (v *view) availableShards() *roaring.Bitmap {
v.mu.RLock()
@ -254,7 +263,7 @@ func (v *view) CreateFragmentIfNotExists(shard uint64) (*fragment, error) {
}
func (v *view) newFragment(path string, shard uint64) *fragment {
frag := newFragment(path, v.index, v.field, v.name, shard)
frag := newFragment(path, v.index, v.field, v.name, shard, v.flags())
frag.CacheType = v.cacheType
frag.CacheSize = v.cacheSize
frag.Logger = v.logger
@ -331,7 +340,7 @@ func (v *view) clearBit(rowID, columnID uint64) (changed bool, err error) {
}
// value uses a column of bits to read a multi-bit value.
func (v *view) value(columnID uint64, bitDepth uint) (value uint64, exists bool, err error) {
func (v *view) value(columnID uint64, bitDepth uint) (value int64, exists bool, err error) {
shard := columnID / ShardWidth
frag, err := v.CreateFragmentIfNotExists(shard)
if err != nil {
@ -341,7 +350,7 @@ func (v *view) value(columnID uint64, bitDepth uint) (value uint64, exists bool,
}
// setValue uses a column of bits to set a multi-bit value.
func (v *view) setValue(columnID uint64, bitDepth uint, value uint64) (changed bool, err error) {
func (v *view) setValue(columnID uint64, bitDepth uint, value int64) (changed bool, err error) {
shard := columnID / ShardWidth
frag, err := v.CreateFragmentIfNotExists(shard)
if err != nil {
@ -351,7 +360,7 @@ func (v *view) setValue(columnID uint64, bitDepth uint, value uint64) (changed b
}
// sum returns the sum & count of a field.
func (v *view) sum(filter *Row, bitDepth uint) (sum, count uint64, err error) {
func (v *view) sum(filter *Row, bitDepth uint) (sum int64, count uint64, err error) {
for _, f := range v.allFragments() {
fsum, fcount, err := f.sum(filter, bitDepth)
if err != nil {
@ -364,7 +373,7 @@ func (v *view) sum(filter *Row, bitDepth uint) (sum, count uint64, err error) {
}
// min returns the min and count of a field.
func (v *view) min(filter *Row, bitDepth uint) (min, count uint64, err error) {
func (v *view) min(filter *Row, bitDepth uint) (min int64, count uint64, err error) {
var minHasValue bool
for _, f := range v.allFragments() {
fmin, fcount, err := f.min(filter, bitDepth)
@ -392,7 +401,7 @@ func (v *view) min(filter *Row, bitDepth uint) (min, count uint64, err error) {
}
// max returns the max and count of a field.
func (v *view) max(filter *Row, bitDepth uint) (max, count uint64, err error) {
func (v *view) max(filter *Row, bitDepth uint) (max int64, count uint64, err error) {
for _, f := range v.allFragments() {
fmax, fcount, err := f.max(filter, bitDepth)
if err != nil {
@ -407,7 +416,7 @@ func (v *view) max(filter *Row, bitDepth uint) (max, count uint64, err error) {
}
// rangeOp returns rows with a field value encoding matching the predicate.
func (v *view) rangeOp(op pql.Token, bitDepth uint, predicate uint64) (*Row, error) {
func (v *view) rangeOp(op pql.Token, bitDepth uint, predicate int64) (*Row, error) {
r := NewRow()
for _, frag := range v.allFragments() {
other, err := frag.rangeOp(op, bitDepth, predicate)
@ -419,6 +428,29 @@ func (v *view) rangeOp(op pql.Token, bitDepth uint, predicate uint64) (*Row, err
return r, nil
}
// upgradeViewBSIv2 upgrades the fragments of v. Returns ok true if any fragment upgraded.
func upgradeViewBSIv2(v *view, bitDepth uint) (ok bool, _ error) {
// If reading from an old formatted BSI roaring bitmap, upgrade and reload.
for _, frag := range v.allFragments() {
if frag.storage.Flags&roaringFlagBSIv2 == 1 {
continue // already upgraded, skip
}
ok = true // mark as upgraded, requires reload
oldPath := frag.path
if newPath, err := upgradeRoaringBSIv2(frag, bitDepth); err != nil {
return ok, errors.Wrap(err, "upgrading bsi v2")
} else if err := frag.closeStorage(); err != nil {
return ok, errors.Wrap(err, "closing after bsi v2 upgrade")
} else if err := os.Rename(oldPath, newPath); err != nil {
return ok, errors.Wrap(err, "renaming after bsi v2 upgrade")
} else if err := frag.openStorage(); err != nil {
return ok, errors.Wrap(err, "re-opening after bsi v2 upgrade")
}
}
return ok, nil
}
// ViewInfo represents schema information for a view.
type ViewInfo struct {
Name string `json:"name"`