scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 09:05:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1448 from travisturner/disco-no-metadata-base

Browse source 
Disco no metadata base
This commit is contained in:
Travis Turner 2021-02-23 10:15:52 -06:00 • committed by GitHub
commit aa43367f45
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
19 changed files with 329 additions and 581 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
api.go
View file

@ -222,7 +222,7 @@ func (api *API) CreateIndex(ctx context.Context, indexName string, options Index
cim := &CreateIndexMessage{
Index: indexName,
CreatedAt: timestamp(),
Meta: &options,
Meta: options,
}
// Create index.

4
broadcast.go
View file

@ -32,10 +32,10 @@ var NopSerializer Serializer = &nopSerializer{}
type nopSerializer struct{}
// Marshal A no-op implementation of Serializer Marshall method.
// Marshal is a no-op implementation of Serializer Marshal method.
func (*nopSerializer) Marshal(Message) ([]byte, error) { return nil, nil }
// Unmarshal A no-op implementation of Serializer Unmarshal method.
// Unmarshal is a no-op implementation of Serializer Unmarshal method.
func (*nopSerializer) Unmarshal([]byte, Message) error { return nil }
// broadcaster is an interface for broadcasting messages.

2
cluster.go
View file

@ -1975,7 +1975,7 @@ type CreateShardMessage struct {
type CreateIndexMessage struct {
Index string
CreatedAt int64
Meta *IndexOptions
Meta IndexOptions
}
// DeleteIndexMessage is an internal message indicating index deletion.

28
dbshard.go
View file

@ -255,7 +255,7 @@ func newIndex2Shards() (r map[txtype]map[string]*shardSet) {
}
type shardSet struct {
shards map[uint64]bool
shardsMap map[uint64]bool
shardsVer int64 // increment with each change.
// give out readonly to repeated consumers if
@ -272,11 +272,11 @@ func (a *shardSet) unionInPlace(b *shardSet) {
}
func (a *shardSet) equals(b *shardSet) bool {
if len(a.shards) != len(b.shards) {
if len(a.shardsMap) != len(b.shardsMap) {
return false
}
for shardInA := range a.shards {
_, ok := b.shards[shardInA]
for shardInA := range a.shardsMap {
_, ok := b.shardsMap[shardInA]
if !ok {
return false
}
@ -285,9 +285,17 @@ func (a *shardSet) equals(b *shardSet) bool {
}
func (a *shardSet) shards() []uint64 {
s := make([]uint64, 0, len(a.shardsMap))
for si := range a.shardsMap {
s = append(s, si)
}
return s
}
func (ss *shardSet) String() (r string) {
r = "["
for k := range ss.shards {
for k := range ss.shardsMap {
r += fmt.Sprintf("%v, ", k)
}
r += "]"
@ -295,9 +303,9 @@ func (ss *shardSet) String() (r string) {
}
func (ss *shardSet) add(shard uint64) {
_, already := ss.shards[shard]
_, already := ss.shardsMap[shard]
if !already {
ss.shards[shard] = true
ss.shardsMap[shard] = true
ss.shardsVer++
}
}
@ -318,7 +326,7 @@ func (ss *shardSet) CloneMaybe() map[uint64]bool {
// must make a fully new copy here.
ss.readonly = make(map[uint64]bool)
for k, v := range ss.shards {
for k, v := range ss.shardsMap {
ss.readonly[k] = v
}
ss.readonlyVer = ss.shardsVer
@ -327,12 +335,12 @@ func (ss *shardSet) CloneMaybe() map[uint64]bool {
func newShardSet() *shardSet {
return &shardSet{
shards: make(map[uint64]bool),
shardsMap: make(map[uint64]bool),
}
}
func newShardSetFromMap(m map[uint64]bool) *shardSet {
return &shardSet{
shards: m,
shardsMap: m,
shardsVer: 1,
}
}

2
dbshard_internal_test.go
View file

@ -334,7 +334,7 @@ func Test_DBPerShard_GetFieldView2Shards_map_from_RBF(t *testing.T) {
cim := &CreateIndexMessage{
Index: index,
CreatedAt: 0,
Meta: &IndexOptions{},
Meta: IndexOptions{},
}
idx, err := holder.createIndex(cim, false)

6
encoding/proto/proto.go
View file

@ -743,7 +743,7 @@ func (s Serializer) encodeCreateIndexMessage(m *pilosa.CreateIndexMessage) *inte
return &internal.CreateIndexMessage{
Index: m.Index,
CreatedAt: m.CreatedAt,
Meta: s.encodeIndexMeta(m.Meta),
Meta: s.encodeIndexMeta(&m.Meta),
}
}
@ -1096,8 +1096,8 @@ func (s Serializer) decodeCreateShardMessage(pb *internal.CreateShardMessage, m
func (s Serializer) decodeCreateIndexMessage(pb *internal.CreateIndexMessage, m *pilosa.CreateIndexMessage) {
m.Index = pb.Index
m.CreatedAt = pb.CreatedAt
m.Meta = &pilosa.IndexOptions{}
s.decodeIndexMeta(pb.Meta, m.Meta)
m.Meta = pilosa.IndexOptions{}
s.decodeIndexMeta(pb.Meta, &m.Meta)
}
func (s Serializer) decodeIndexMeta(pb *internal.IndexMeta, m *pilosa.IndexOptions) {

300
field.go
View file

@ -30,9 +30,7 @@ import (
"sync"
"time"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/v2/disco"
"github.com/pilosa/pilosa/v2/internal"
"github.com/pilosa/pilosa/v2/pql"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/stats"
@ -109,15 +107,6 @@ type Field struct {
// Field options.
options FieldOptions
// finalOptions is used with a final call to applyOptions.
// The initial call to applyOptions is made with options
// loaded from the meta file on disk (in the case when
// a field is being re-opened). If the field creator calls
// setOptions before calling Open(), then those options
// will be held in finalOptions, and applied instead of
// those from the meta file.
finalOptions *FieldOptions
bsiGroups []*bsiGroup
// Shards with data on any node in the cluster, according to this node.
@ -373,7 +362,7 @@ func newField(holder *Holder, path, index, name string, opts FieldOption) (*Fiel
schemator: disco.NopSchemator,
serializer: NopSerializer,
options: *applyDefaultOptions(&fo),
options: applyDefaultOptions(&fo),
remoteAvailableShards: roaring.NewBitmap(),
@ -543,26 +532,6 @@ func (f *Field) Type() string {
return f.options.Type
}
// SetCacheSize sets the cache size for ranked fames. Persists to meta file on update.
// defaults to DefaultCacheSize 50000
func (f *Field) SetCacheSize(v uint32) error {
f.mu.Lock()
defer f.mu.Unlock()
// Ignore if no change occurred.
if v == 0 || f.options.CacheSize == v {
return nil
}
// Persist meta data to disk on change.
f.options.CacheSize = v
if err := f.saveMeta(); err != nil {
return errors.Wrap(err, "saving")
}
return nil
}
// CacheSize returns the ranked field cache size.
func (f *Field) CacheSize() uint32 {
f.mu.RLock()
@ -587,24 +556,12 @@ func (f *Field) Open() error {
return errors.Wrap(err, "creating field dir")
}
f.holder.Logger.Debugf("load meta file for index/field: %s/%s", f.index, f.name)
if err := f.loadMeta(); err != nil {
return errors.Wrap(err, "loading meta")
}
f.holder.Logger.Debugf("load available shards for index/field: %s/%s", f.index, f.name)
if err := f.loadAvailableShards(); err != nil {
return errors.Wrap(err, "loading available shards")
}
// If options were provided using setOptions(), then
// use those instead of the options from the meta file.
if f.finalOptions != nil {
f.options = *f.finalOptions
}
// Apply the field options loaded from meta (or set via setOptions()).
// Apply the field options loaded from etcd (or set via setOptions()).
f.holder.Logger.Debugf("apply options for index/field: %s/%s", f.index, f.name)
if err := f.applyOptions(f.options); err != nil {
return errors.Wrap(err, "applying options")
@ -632,6 +589,7 @@ func (f *Field) Open() error {
return errors.Wrap(err, "checking foreign index")
}
}
f.availableShardChan = make(chan []byte)
f.doneChan = make(chan struct{})
f.wg.Add(1)
@ -750,6 +708,28 @@ func (f *Field) ForeignIndex() string {
return f.options.ForeignIndex
}
func (f *Field) bitDepth() (uint64, error) {
var maxBitDepth uint64
view2shards := f.idx.fieldView2shard.getViewsForField(f.name)
for name, shardset := range view2shards {
view := f.view(name)
if view == nil {
continue
}
bd, err := view.bitDepth(shardset.shards())
if err != nil {
return 0, errors.Wrapf(err, "getting view(%s) bit depth", name)
}
if bd > maxBitDepth {
maxBitDepth = bd
}
}
return maxBitDepth, nil
}
// openViews opens and initializes the views inside the field.
func (f *Field) openViews() error {
view2shards := f.idx.fieldView2shard.getViewsForField(f.name)
@ -759,29 +739,11 @@ func (f *Field) openViews() error {
}
for name, shardset := range view2shards {
view := f.newView(f.viewPath(name), name)
if err := view.openWithShardSet(shardset); err != nil {
return fmt.Errorf("opening view: view=%s, err=%s", view.name, err)
}
if f.holder.txf.TxType() == RoaringTxn {
// 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.openWithShardSet(shardset); err != nil {
return fmt.Errorf("re-opening view: view=%s, err=%s", view.name, err)
}
}
}
}
view.rowAttrStore = f.rowAttrStore
f.holder.Logger.Debugf("add index/field/view to field.viewMap: %s/%s/%s", f.index, f.name, view.name)
f.viewMap[view.name] = view
@ -789,98 +751,9 @@ func (f *Field) openViews() error {
return nil
}
// loadMeta reads meta data for the field, if any.
func (f *Field) loadMeta() error {
var pb internal.FieldOptions
// Read data from meta file.
buf, err := ioutil.ReadFile(filepath.Join(f.path, ".meta"))
if os.IsNotExist(err) {
return nil
} else if err != nil {
return errors.Wrap(err, "reading meta")
} else {
if err := proto.Unmarshal(buf, &pb); err != nil {
return errors.Wrap(err, "unmarshaling")
}
}
// Since pb.Min and pb.Max were changed to pql.Decimal,
// and since they now have a different protobuf field
// number, an existing meta file may have values in the
// old min/max fields which need to be converted to
// pql.Decimal.
// TODO: we can remove the OldMin/OldMax once we're
// confident no one is still using the older version.
var min pql.Decimal
if pb.Min != nil {
min = pql.NewDecimal(pb.Min.Value, pb.Min.Scale)
} else {
min = pql.NewDecimal(pb.OldMin, pb.Scale)
}
var max pql.Decimal
if pb.Max != nil {
max = pql.NewDecimal(pb.Max.Value, pb.Max.Scale)
} else {
max = pql.NewDecimal(pb.OldMax, pb.Scale)
}
// Initialize "base" to "min" when upgrading from v1 BSI format.
if pb.BitDepth == 0 {
minInt64, maxInt64 := min.ToInt64(0), max.ToInt64(0)
pb.Base = bsiBase(minInt64, maxInt64)
pb.BitDepth = uint64(bitDepthInt64(maxInt64 - minInt64))
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 = min
f.options.Max = max
f.options.Base = pb.Base
f.options.Scale = pb.Scale
f.options.BitDepth = pb.BitDepth
f.options.TimeQuantum = TimeQuantum(pb.TimeQuantum)
f.options.Keys = pb.Keys
f.options.NoStandardView = pb.NoStandardView
f.options.ForeignIndex = pb.ForeignIndex
return nil
}
// saveMeta writes meta data for the field.
func (f *Field) saveMeta() error {
path := filepath.Join(f.path, ".meta")
// Create a temporary file to marshal to.
tempPath := f.path + tempExt
// Marshal metadata.
fo := f.options
buf, err := proto.Marshal(fo.encode())
if err != nil {
return errors.Wrap(err, "marshaling")
}
// Write to meta file.
if err := ioutil.WriteFile(tempPath, buf, 0666); err != nil {
return errors.Wrap(err, "writing meta")
}
// Move temp file to data file location.
if err := os.Rename(tempPath, path); err != nil {
return fmt.Errorf("rename temp: %s", err)
}
return nil
}
// setOptions saves options for final application during Open().
func (f *Field) setOptions(opts *FieldOptions) {
f.finalOptions = applyDefaultOptions(opts)
f.options = applyDefaultOptions(opts)
}
// applyOptions configures the field based on opt.
@ -930,10 +803,7 @@ func (f *Field) applyOptions(opt FieldOptions) error {
Scale: opt.Scale,
BitDepth: opt.BitDepth,
}
// Validate bsiGroup.
if err := bsig.validate(); err != nil {
return err
}
// Validate and create bsiGroup.
if err := f.createBSIGroup(bsig); err != nil {
return errors.Wrap(err, "creating bsigroup")
}
@ -947,11 +817,11 @@ func (f *Field) applyOptions(opt FieldOptions) error {
f.options.BitDepth = 0
f.options.Keys = opt.Keys
f.options.NoStandardView = opt.NoStandardView
// Set the time quantum.
if err := f.setTimeQuantum(opt.TimeQuantum); err != nil {
f.Close()
return errors.Wrap(err, "setting time quantum")
// Validate the time quantum.
if !opt.TimeQuantum.Valid() {
return ErrInvalidTimeQuantum
}
f.options.TimeQuantum = opt.TimeQuantum
f.options.ForeignIndex = opt.ForeignIndex
case FieldTypeBool:
f.options.Type = FieldTypeBool
@ -1044,17 +914,6 @@ func (f *Field) createBSIGroup(bsig *bsiGroup) error {
defer f.mu.Unlock()
// Append bsiGroup.
if err := f.addBSIGroup(bsig); err != nil {
return err
}
if err := f.saveMeta(); err != nil {
return errors.Wrap(err, "saving")
}
return nil
}
// addBSIGroup adds a single bsiGroup to bsiGroups.
func (f *Field) addBSIGroup(bsig *bsiGroup) error {
if err := bsig.validate(); err != nil {
return errors.Wrap(err, "validating bsigroup")
} else if f.hasBSIGroup(bsig.Name) {
@ -1079,27 +938,6 @@ func (f *Field) TimeQuantum() TimeQuantum {
return f.options.TimeQuantum
}
// setTimeQuantum sets the time quantum for the field.
func (f *Field) setTimeQuantum(q TimeQuantum) error {
f.mu.Lock()
defer f.mu.Unlock()
// Validate input.
if !q.Valid() {
return ErrInvalidTimeQuantum
}
// Update value on field.
f.options.TimeQuantum = q
// Persist meta data to disk.
if err := f.saveMeta(); err != nil {
return errors.Wrap(err, "saving meta")
}
return nil
}
// RowTime gets the row at the particular time with the granularity specified by
// the quantum.
func (f *Field) RowTime(tx Tx, rowID uint64, time time.Time, quantum string) (*Row, error) {
@ -1448,22 +1286,16 @@ func (f *Field) SetValue(tx Tx, columnID uint64, value int64) (changed bool, err
// 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")
uvalue := uint64(baseValue)
if value < 0 {
uvalue = uint64(-baseValue)
}
bitDepth := bitDepth(uvalue)
f.mu.Lock()
bsig.BitDepth = bitDepth
f.options.BitDepth = bitDepth
f.mu.Unlock()
}
// Fetch target view.
@ -1764,20 +1596,14 @@ func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, option
requiredDepth = v
}
// Increase bit depth if required.
if err := func() error {
bitDepth := bsig.BitDepth
if requiredDepth > bitDepth {
f.mu.Lock()
defer f.mu.Unlock()
bitDepth := bsig.BitDepth
if requiredDepth > bitDepth {
bsig.BitDepth = requiredDepth
f.options.BitDepth = requiredDepth
return f.saveMeta()
} else {
requiredDepth = bitDepth
}
return nil
}(); err != nil {
return errors.Wrap(err, "increasing bsi bit depth")
bsig.BitDepth = requiredDepth
f.options.BitDepth = requiredDepth
f.mu.Unlock()
} else {
requiredDepth = bitDepth
}
// Import into each fragment.
@ -1959,38 +1785,16 @@ func newFieldOptions(opts ...FieldOption) (*FieldOptions, error) {
// applyDefaultOptions updates FieldOptions with the default
// values if o does not contain a valid type.
func applyDefaultOptions(o *FieldOptions) *FieldOptions {
func applyDefaultOptions(o *FieldOptions) FieldOptions {
if o == nil {
o = &FieldOptions{}
}
if o.Type == "" {
o.Type = DefaultFieldType
o.CacheType = DefaultCacheType
o.CacheSize = DefaultCacheSize
}
return o
}
// encode converts o into its internal representation.
func (o *FieldOptions) encode() *internal.FieldOptions {
return encodeFieldOptions(o)
}
func encodeFieldOptions(o *FieldOptions) *internal.FieldOptions {
if o == nil {
return nil
}
return &internal.FieldOptions{
Type: o.Type,
CacheType: o.CacheType,
CacheSize: o.CacheSize,
Base: o.Base,
Scale: o.Scale,
BitDepth: uint64(o.BitDepth),
Min: &internal.Decimal{Value: o.Min.Value, Scale: o.Min.Scale},
Max: &internal.Decimal{Value: o.Max.Value, Scale: o.Max.Scale},
TimeQuantum: string(o.TimeQuantum),
Keys: o.Keys,
NoStandardView: o.NoStandardView,
ForeignIndex: o.ForeignIndex,
}
return *o
}
// MarshalJSON marshals FieldOptions to JSON such that

59
field_internal_test.go
View file

@ -15,6 +15,7 @@
package pilosa
import (
"context"
"fmt"
"math"
"os"
@ -207,7 +208,8 @@ func NewTestField(t *testing.T, opts FieldOption) *TestField {
if err != nil {
t.Fatal(err)
}
h := NewHolder(path, nil)
h := NewHolder(path, DefaultHolderConfig())
panicOn(h.Open())
idx, err := h.CreateIndex("i", IndexOptions{})
@ -247,7 +249,11 @@ func (f *TestField) Reopen() error {
f.parent = nil
return err
}
if err := f.parent.Open(); err != nil {
schema, err := f.parent.Schemator.Schema(context.Background())
if err != nil {
return err
}
if err := f.parent.OpenWithSchema(schema[f.parent.name]); err != nil {
f.parent = nil
return err
}
@ -297,13 +303,11 @@ func TestField_CreateViewIfNotExists(t *testing.T) {
}
func TestField_SetTimeQuantum(t *testing.T) {
f := OpenField(t, OptFieldTypeTime(TimeQuantum("")))
f := OpenField(t, OptFieldTypeTime(TimeQuantum("YMDH")))
defer f.Close()
// Set & retrieve time quantum.
if err := f.setTimeQuantum(TimeQuantum("YMDH")); err != nil {
t.Fatal(err)
} else if q := f.TimeQuantum(); q != TimeQuantum("YMDH") {
// Retrieve time quantum.
if q := f.TimeQuantum(); q != TimeQuantum("YMDH") {
t.Fatalf("unexpected quantum: %s", q)
}
@ -316,17 +320,13 @@ func TestField_SetTimeQuantum(t *testing.T) {
}
func TestField_RowTime(t *testing.T) {
f := OpenField(t, OptFieldTypeTime(TimeQuantum("")))
f := OpenField(t, OptFieldTypeTime(TimeQuantum("YMDH")))
defer f.Close()
// Obtain transaction.
tx := f.idx.holder.txf.NewTx(Txo{Write: writable, Index: f.idx, Field: f.Field, Shard: 0})
defer tx.Rollback()
if err := f.setTimeQuantum(TimeQuantum("YMDH")); err != nil {
t.Fatal(err)
}
f.MustSetBit(tx, 1, 1, time.Date(2010, time.January, 5, 12, 0, 0, 0, time.UTC))
f.MustSetBit(tx, 1, 2, time.Date(2011, time.January, 5, 12, 0, 0, 0, time.UTC))
f.MustSetBit(tx, 1, 3, time.Date(2010, time.February, 5, 12, 0, 0, 0, time.UTC))
@ -552,7 +552,7 @@ func TestField_ApplyOptions(t *testing.T) {
} {
fld := &Field{}
fld.options = *applyDefaultOptions(&FieldOptions{})
fld.options = applyDefaultOptions(&FieldOptions{})
if err := fld.applyOptions(tt.opts); err != nil {
t.Fatal(err)
@ -922,3 +922,36 @@ func TestBSIGroup_TxReopenDB(t *testing.T) {
// the test: can we re-open a BSI fragment under Tx store
_ = f.Reopen()
}
// Ensure that an integer field has the same BitDepth after reopening.
func TestField_SaveMeta(t *testing.T) {
f := OpenField(t, OptFieldTypeInt(-10, 1000))
defer f.Close()
colID := uint64(1)
val := int64(88)
expBitDepth := uint64(7)
// Obtain transaction.
tx := f.idx.holder.txf.NewTx(Txo{Write: writable, Index: f.idx, Field: f.Field, Shard: 0})
defer tx.Rollback()
if changed, err := f.SetValue(tx, colID, val); err != nil {
t.Fatal(err)
} else if !changed {
t.Fatal("expected SetValue to return changed = true")
}
if f.options.BitDepth != expBitDepth {
t.Fatalf("expected BitDepth after set to be: %d, got: %d", expBitDepth, f.options.BitDepth)
}
// Reload field and verify that it is persisted.
if err := f.Reopen(); err != nil {
t.Fatal(err)
}
if f.options.BitDepth != expBitDepth {
t.Fatalf("expected BitDepth after reopen to be: %d, got: %d", expBitDepth, f.options.BitDepth)
}
}

64
fragment.go
View file

@ -238,6 +238,24 @@ func newFragment(holder *Holder, spec fragSpec, shard uint64, flags byte) *fragm
// cachePath returns the path to the fragment's cache data.
func (f *fragment) cachePath() string { return f.path() + cacheExt }
func (f *fragment) bitDepth() (uint64, error) {
tx, err := f.holder.BeginTx(false, f.idx, f.shard)
if err != nil {
return 0, errors.Wrapf(err, "beginning new tx(false, %s, %d)", f.index(), f.shard)
}
defer tx.Rollback()
maxRowID, _, err := f.maxRow(tx, nil)
if err != nil {
return 0, errors.Wrapf(err, "getting fragment max row id")
}
if maxRowID+1 > bsiOffsetBit {
return maxRowID + 1 - bsiOffsetBit, nil
}
return 0, nil
}
type FragmentInfo struct {
BitmapInfo roaring.BitmapInfo
BlockChecksums []FragmentBlock `json:"BlockChecksums,omitempty"`
@ -3279,52 +3297,6 @@ func (f *fragment) blockToRoaringData(block int) ([]byte, error) {
})
}
// 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 uint64) (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) error {
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
}
return nil
})
}()
// 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 interface {
// TODO(kuba) linter suggests to use io.Seeker
// Seek(offset int64, whence int) (int64, error)

2
fragment_internal_test.go
View file

@ -3588,7 +3588,7 @@ func fragTestMustOpenIndex(index string, holder *Holder, opt IndexOptions) *Inde
cim := &CreateIndexMessage{
Index: index,
CreatedAt: 0,
Meta: &opt,
Meta: opt,
}
holder.mu.Lock()

40
holder.go
View file

@ -234,7 +234,7 @@ func DefaultHolderConfig() *HolderConfig {
OpenTransactionStore: OpenInMemTransactionStore,
OpenIDAllocator: func(string) (*idAllocator, error) { return &idAllocator{}, nil },
TranslationSyncer: NopTranslationSyncer,
Serializer: NopSerializer,
Serializer: GobSerializer,
Schemator: disco.InMemSchemator,
CacheFlushInterval: defaultCacheFlushInterval,
StatsClient: stats.NopStatsClient,
@ -652,7 +652,7 @@ func (h *Holder) Open() error {
// decode the CreateIndexMessage from the schema data in order to
// get its metadata, such as CreateAt.
cim, err := h.decodeCreateIndexMessage(idx.Data)
cim, err := decodeCreateIndexMessage(h.serializer, idx.Data)
if err != nil {
return errors.Wrap(err, "decoding create index message")
}
@ -885,7 +885,7 @@ func (h *Holder) schema(ctx context.Context, includeViews bool) ([]*IndexInfo, e
}
for _, index := range schema {
cim, err := h.decodeCreateIndexMessage(index.Data)
cim, err := decodeCreateIndexMessage(h.serializer, index.Data)
if err != nil {
return nil, errors.Wrap(err, "decoding CreateIndexMessage")
}
@ -893,7 +893,7 @@ func (h *Holder) schema(ctx context.Context, includeViews bool) ([]*IndexInfo, e
di := &IndexInfo{
Name: cim.Index,
CreatedAt: cim.CreatedAt,
Options: *cim.Meta,
Options: cim.Meta,
ShardWidth: ShardWidth,
Fields: make([]*FieldInfo, 0, len(index.Fields)),
}
@ -901,7 +901,7 @@ func (h *Holder) schema(ctx context.Context, includeViews bool) ([]*IndexInfo, e
if fieldName == existenceFieldName {
continue
}
cfm, err := h.decodeCreateFieldMessage(field.Data)
cfm, err := decodeCreateFieldMessage(h.serializer, field.Data)
if err != nil {
return nil, errors.Wrap(err, "decoding CreateFieldMessage")
}
@ -1013,7 +1013,7 @@ func (h *Holder) CreateIndex(name string, opt IndexOptions) (*Index, error) {
cim := &CreateIndexMessage{
Index: name,
CreatedAt: timestamp(),
Meta: &opt,
Meta: opt,
}
// Create the index in etcd as the system of record.
@ -1107,7 +1107,7 @@ func (h *Holder) CreateIndexIfNotExists(name string, opt IndexOptions) (*Index,
cim := &CreateIndexMessage{
Index: name,
CreatedAt: timestamp(),
Meta: &opt,
Meta: opt,
}
// Create the index in etcd as the system of record.
@ -1148,27 +1148,19 @@ func (h *Holder) createIndex(cim *CreateIndexMessage, broadcast bool) (*Index, e
return nil, errors.New("index name required")
}
opt := cim.Meta
if opt == nil {
opt = &IndexOptions{}
}
// Otherwise create a new index.
index, err := h.newIndex(h.IndexPath(cim.Index), cim.Index)
if err != nil {
return nil, errors.Wrap(err, "creating")
}
index.keys = opt.Keys
index.trackExistence = opt.TrackExistence
index.keys = cim.Meta.Keys
index.trackExistence = cim.Meta.TrackExistence
index.createdAt = cim.CreatedAt
if err = index.Open(); err != nil {
return nil, errors.Wrap(err, "opening")
}
if err = index.saveMeta(); err != nil {
return nil, errors.Wrap(err, "meta")
}
// Update options.
h.addIndex(index)
@ -1231,7 +1223,7 @@ func (h *Holder) loadIndex(indexName string) (*Index, error) {
return nil, errors.Wrapf(err, "getting index: %s", indexName)
}
cim, err := h.decodeCreateIndexMessage(b)
cim, err := decodeCreateIndexMessage(h.serializer, b)
if err != nil {
return nil, errors.Wrap(err, "decoding CreateIndexMessage")
}
@ -1251,7 +1243,7 @@ func (h *Holder) loadField(indexName, fieldName string) (*Field, error) {
return nil, errors.Errorf("local index not found: %s", indexName)
}
cfm, err := h.decodeCreateFieldMessage(b)
cfm, err := decodeCreateFieldMessage(h.serializer, b)
if err != nil {
return nil, errors.Wrap(err, "decoding CreateFieldMessage")
}
@ -1284,7 +1276,7 @@ func (h *Holder) newIndex(path, name string) (*Index, error) {
index.Stats = h.Stats.WithTags(fmt.Sprintf("index:%s", index.Name()))
index.broadcaster = h.broadcaster
index.serializer = h.serializer
index.schemator = h.schemator
index.Schemator = h.schemator
index.newAttrStore = h.NewAttrStore
index.columnAttrs = h.NewAttrStore(filepath.Join(index.path, ".data"))
index.OpenTranslateStore = h.OpenTranslateStore
@ -2284,17 +2276,17 @@ func (h *Holder) HasRoaringData() (has bool, err error) {
return
}
func (h *Holder) decodeCreateIndexMessage(b []byte) (*CreateIndexMessage, error) {
func decodeCreateIndexMessage(ser Serializer, b []byte) (*CreateIndexMessage, error) {
var cim CreateIndexMessage
if err := h.serializer.Unmarshal(b, &cim); err != nil {
if err := ser.Unmarshal(b, &cim); err != nil {
return nil, errors.Wrap(err, "unmarshaling")
}
return &cim, nil
}
func (h *Holder) decodeCreateFieldMessage(b []byte) (*CreateFieldMessage, error) {
func decodeCreateFieldMessage(ser Serializer, b []byte) (*CreateFieldMessage, error) {
var cfm CreateFieldMessage
if err := h.serializer.Unmarshal(b, &cfm); err != nil {
if err := ser.Unmarshal(b, &cfm); err != nil {
return nil, errors.Wrap(err, "unmarshaling")
}
return &cfm, nil

195
index.go
View file

@ -17,7 +17,6 @@ package pilosa
import (
"context"
"fmt"
"io/ioutil"
"os"
"path/filepath"
"sort"
@ -25,9 +24,7 @@ import (
"sync"
"time"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/v2/disco"
"github.com/pilosa/pilosa/v2/internal"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/stats"
"github.com/pilosa/pilosa/v2/testhook"
@ -57,7 +54,7 @@ type Index struct {
columnAttrs AttrStore
broadcaster broadcaster
schemator disco.Schemator
Schemator disco.Schemator
serializer Serializer
Stats stats.StatsClient
@ -102,7 +99,7 @@ func NewIndex(holder *Holder, path, name string) (*Index, error) {
holder: holder,
trackExistence: true,
schemator: disco.InMemSchemator,
Schemator: disco.InMemSchemator,
serializer: NopSerializer,
translateStores: make(map[int]TranslateStore),
@ -183,9 +180,28 @@ func (i *Index) Open() error {
// OpenWithSchema opens the index and uses the provided schema to verify that
// the index's fields are expected.
func (i *Index) OpenWithSchema(idx *disco.Index) error {
if idx == nil {
return ErrInvalidSchema
}
// decode the CreateIndexMessage from the schema data in order to
// get its metadata.
cim, err := decodeCreateIndexMessage(i.serializer, idx.Data)
if err != nil {
return errors.Wrap(err, "decoding create index message")
}
i.createdAt = cim.CreatedAt
i.trackExistence = cim.Meta.TrackExistence
i.keys = cim.Meta.Keys
return i.open(idx)
}
// open opens the index with an optional schema (disco.Index). If a schema is
// provided, it will apply the metadata from the schema to the index, and then
// open all fields found in the schema. If a schema is not provided, the
// metadata for the index is not changed from its existing value, and fields are
// not validated against the schema as they are opened.
func (i *Index) open(idx *disco.Index) (err error) {
// Ensure the path exists.
i.holder.Logger.Debugf("ensure index path exists: %s", i.path)
@ -193,12 +209,6 @@ func (i *Index) open(idx *disco.Index) (err error) {
return errors.Wrap(err, "creating directory")
}
// Read meta file.
i.holder.Logger.Debugf("load meta file for index: %s", i.name)
if err := i.loadMeta(); err != nil {
return errors.Wrap(err, "loading meta file")
}
// we don't want to open *all* the views for each shard, since
// most are empty when we are doing time quantums. It slows
// down startup dramatically. So we ask for the meta data
@ -209,11 +219,27 @@ func (i *Index) open(idx *disco.Index) (err error) {
}
i.fieldView2shard = fieldView2shard
// Add index to a map in holder. Used by openFields.
i.holder.addIndex(i)
i.holder.Logger.Debugf("open fields for index: %s", i.name)
if err := i.openFields(idx); err != nil {
return errors.Wrap(err, "opening fields")
}
// Set bit depths.
// This is called in Index.open() (as opposed to Field.Open()) because the
// Field.bitDepth() method uses a transaction which relies on the index and
// its entry for the field in the Index.field map. If we try to set a
// field's BitDepth in Field.Open(), which itself might be inside the
// Index.openField() loop, then the field has not yet been added to the
// Index.field map. I think it would be better if Field.bitDepth didn't rely
// on its index at all, but perhaps with transactions that not possible. I
// don't know.
if err := i.setFieldBitDepths(); err != nil {
return errors.Wrap(err, "setting field bitDepths")
}
if i.trackExistence {
if err := i.openExistenceField(); err != nil {
return errors.Wrap(err, "opening existence field")
@ -273,11 +299,6 @@ func (i *Index) openFields(idx *disco.Index) error {
eg, ctx := errgroup.WithContext(context.Background())
var mu sync.Mutex
// var flds map[string]*disco.Field
// if idx != nil {
// flds = idx.Fields
// }
fileLoop:
for _, loopFi := range fis {
select {
@ -293,34 +314,25 @@ fileLoop:
continue
}
var createdAt int64
var cfm *CreateFieldMessage = &CreateFieldMessage{}
var err error
// Only continue with indexes which are present in the provided,
// Only continue with fields which are present in the provided,
// non-nil index schema. The reason we have to check for idx != nil
// here is because there are tests which call index.Open on an index
// with a NopSchemator. A better approach might be for those tests
// to use a mock Schemator which returns a schema containing the
// index. For an example, see TestField_SetTimeQuantum which
// re-opens a field and curiously has to re-open that field's index
// because at some point we introduced a pointer from the field back
// to its index (possibly related to transactions?).
// here is because there are tests which call index.Open without
// having a disco.Index available.
if idx != nil {
fld, ok := idx.Fields[fi.Name()]
//fld, ok := flds[fi.Name()]
if !ok {
continue
}
// decode the CreateIndexMessage from the schema data in order to
// get its metadata, such as CreateAt.
// TODO: similar to the createdAt TODO in holder, it may no
// longer be necessary to keep createdAt on the in-memory field
// struct.
cfm, err := i.holder.decodeCreateFieldMessage(fld.Data)
// Decode the CreateFieldMessage from the schema data in order to
// get its metadata.
cfm, err = decodeCreateFieldMessage(i.holder.serializer, fld.Data)
if err != nil {
return errors.Wrap(err, "decoding create field message")
}
createdAt = cfm.CreatedAt
}
indexQueue <- struct{}{}
@ -330,7 +342,7 @@ fileLoop:
}()
i.holder.Logger.Debugf("open field: %s", fi.Name())
_, err := i.openField(&mu, createdAt, fi.Name())
_, err := i.openField(&mu, cfm, fi.Name())
if err != nil {
return errors.Wrap(err, "opening field")
}
@ -353,12 +365,8 @@ fileLoop:
// openField opens the field directory, initializes the field, and adds it to
// the in-memory map of fields maintained by Index.
func (i *Index) openField(mu *sync.Mutex, createdAt int64, file string) (*Field, error) {
func (i *Index) openField(mu *sync.Mutex, cfm *CreateFieldMessage, file string) (*Field, error) {
mu.Lock()
// goroutine safe
i.holder.addIndex(i)
fld, err := i.newField(i.fieldPath(filepath.Base(file)), filepath.Base(file))
mu.Unlock()
if err != nil {
@ -369,7 +377,8 @@ func (i *Index) openField(mu *sync.Mutex, createdAt int64, file string) (*Field,
// up a foreign index.
fld.holder = i.holder
fld.createdAt = createdAt
fld.createdAt = cfm.CreatedAt
fld.options = applyDefaultOptions(cfm.Meta)
// open the views we have data for.
if err := fld.Open(); err != nil {
@ -386,10 +395,17 @@ func (i *Index) openField(mu *sync.Mutex, createdAt int64, file string) (*Field,
// openExistenceField gets or creates the existence field and associates it to the index.
func (i *Index) openExistenceField() error {
cfm := &CreateFieldMessage{
Index: i.name,
Field: existenceFieldName,
CreatedAt: 0,
Meta: &FieldOptions{CacheType: CacheTypeNone, CacheSize: 0},
}
// First try opening the existence field from disk. If it doesn't already
// exist on disk, then we fall through to the code path which creates it.
var mu sync.Mutex
fld, err := i.openField(&mu, 0, existenceFieldName)
fld, err := i.openField(&mu, cfm, existenceFieldName)
if err == nil {
i.existenceFld = fld
return nil
@ -399,12 +415,6 @@ func (i *Index) openExistenceField() error {
// If we have gotten here, it means that we couldn't successfully open the
// existence field from disk, so we need to create it.
cfm := &CreateFieldMessage{
Index: i.name,
Field: existenceFieldName,
CreatedAt: 0,
Meta: &FieldOptions{CacheType: CacheTypeNone, CacheSize: 0},
}
f, err := i.createFieldIfNotExists(cfm)
if err != nil {
@ -414,56 +424,28 @@ func (i *Index) openExistenceField() error {
return nil
}
// loadMeta reads meta data for the index, if any.
func (i *Index) loadMeta() error {
// TrackExistence is by default true
pb := &internal.IndexMeta{TrackExistence: true}
// Read data from meta file.
buf, err := ioutil.ReadFile(filepath.Join(i.path, ".meta"))
if os.IsNotExist(err) {
return nil
} else if err != nil {
return errors.Wrap(err, "reading")
} else {
if err := proto.Unmarshal(buf, pb); err != nil {
return errors.Wrap(err, "unmarshalling")
// setFieldBitDepths sets the BitDepth for all int and decimal fields in the index.
func (i *Index) setFieldBitDepths() error {
for name, f := range i.fields {
switch f.Type() {
case FieldTypeInt, FieldTypeDecimal:
// pass
default:
continue
}
bd, err := f.bitDepth()
if err != nil {
return errors.Wrapf(err, "getting bit depth for field: %s", name)
}
f.mu.Lock()
f.options.BitDepth = bd
f.mu.Unlock()
}
// Copy metadata fields.
if pb == nil {
i.trackExistence = true
} else {
i.trackExistence = pb.TrackExistence
}
i.keys = pb.GetKeys()
return nil
}
// saveMeta writes meta data for the index.
func (i *Index) saveMeta() error {
// Marshal metadata.
buf, err := proto.Marshal(&internal.IndexMeta{
Keys: i.keys,
TrackExistence: i.trackExistence,
})
if err != nil {
return errors.Wrap(err, "marshalling")
}
// Write to meta file.
if err := ioutil.WriteFile(filepath.Join(i.path, ".meta"), buf, 0666); err != nil {
return errors.Wrap(err, "writing")
}
return nil
}
// Close closes the index and its fields.
func (i *Index) Close() error {
i.mu.Lock()
defer i.mu.Unlock()
defer func() {
@ -592,7 +574,7 @@ func (i *Index) CreateField(name string, opts ...FieldOption) (*Field, error) {
cfm := &CreateFieldMessage{
Index: i.name,
Field: name,
CreatedAt: 0,
CreatedAt: timestamp(),
Meta: fo,
}
@ -691,7 +673,7 @@ func (i *Index) CreateFieldIfNotExistsWithOptions(name string, opt *FieldOptions
cfm := &CreateFieldMessage{
Index: i.name,
Field: name,
CreatedAt: 0,
CreatedAt: timestamp(),
Meta: opt,
}
@ -721,7 +703,7 @@ func (i *Index) persistField(ctx context.Context, cfm *CreateFieldMessage) error
if b, err := i.serializer.Marshal(cfm); err != nil {
return errors.Wrap(err, "marshaling")
} else if err := i.schemator.CreateField(ctx, cfm.Index, cfm.Field, b); err != nil {
} else if err := i.Schemator.CreateField(ctx, cfm.Index, cfm.Field, b); err != nil {
return errors.Wrapf(err, "writing field to disco: %s/%s", cfm.Index, cfm.Field)
}
return nil
@ -752,6 +734,7 @@ func (i *Index) createField(cfm *CreateFieldMessage, broadcast bool) (*Field, er
opt = &FieldOptions{}
}
// TODO: can we do a general FieldOption validation here instead of just cache type?
if cfm.Field == "" {
return nil, errors.New("field name required")
} else if opt.CacheType != "" && !isValidCacheType(opt.CacheType) {
@ -776,11 +759,6 @@ func (i *Index) createField(cfm *CreateFieldMessage, broadcast bool) (*Field, er
return nil, errors.Wrap(err, "opening")
}
if err := f.saveMeta(); err != nil {
f.Close()
return nil, errors.Wrap(err, "saving meta")
}
// Add to index's field lookup.
i.fields[cfm.Field] = f
@ -810,7 +788,7 @@ func (i *Index) newField(path, name string) (*Field, error) {
f.idx = i
f.Stats = i.Stats
f.broadcaster = i.broadcaster
f.schemator = i.schemator
f.schemator = i.Schemator
f.serializer = i.serializer
f.rowAttrStore = i.newAttrStore(filepath.Join(f.path, ".data"))
f.OpenTranslateStore = i.OpenTranslateStore
@ -822,6 +800,11 @@ func (i *Index) DeleteField(name string) error {
i.mu.Lock()
defer i.mu.Unlock()
// Disallow deleting the existence field.
if name == existenceFieldName {
return newNotFoundError(ErrFieldNotFound, existenceFieldName)
}
// Confirm field exists.
f := i.field(name)
if f == nil {
@ -837,23 +820,11 @@ func (i *Index) DeleteField(name string) error {
return errors.Wrap(err, "Txf.DeleteFieldFromStore")
}
// If the field being deleted is the existence field,
// turn off existence tracking on the index.
if name == existenceFieldName {
i.trackExistence = false
i.existenceFld = nil
// Update meta data on disk.
if err := i.saveMeta(); err != nil {
return errors.Wrap(err, "saving existence meta data")
}
}
// Remove reference.
delete(i.fields, name)
// Delete the field from etcd as the system of record.
if err := i.schemator.DeleteField(context.TODO(), i.name, name); err != nil {
if err := i.Schemator.DeleteField(context.TODO(), i.name, name); err != nil {
return errors.Wrapf(err, "deleting field from etcd: %s/%s", i.name, name)
}

37
index_internal_test.go
View file

@ -52,40 +52,3 @@ func (i *Index) reopen() error {
}
return nil
}
// Ensure that deleting the existence field is handled properly.
func TestIndex_Existence_Delete(t *testing.T) {
// Create Index (with existence tracking).
index := mustOpenIndex(t, IndexOptions{TrackExistence: true})
defer index.Close()
// Ensure existence field has been created.
ef := index.Field(existenceFieldName)
if ef == nil {
t.Fatalf("expected field to have been created: %s", existenceFieldName)
} else if !index.trackExistence {
t.Fatalf("expected index.trackExistence to be true")
} else if index.existenceFld == nil {
t.Fatalf("expected index.existenceField to be non-nil")
}
// Delete existence field.
if err := index.DeleteField(existenceFieldName); err != nil {
t.Fatal(err)
}
// Re-open index.
if err := index.reopen(); err != nil {
t.Fatal(err)
}
// Ensure existence field no longer exists.
ef = index.Field(existenceFieldName)
if ef != nil {
t.Fatalf("expected field to have been deleted: %s", existenceFieldName)
} else if index.trackExistence {
t.Fatalf("expected index.trackExistence to be false")
} else if index.existenceFld != nil {
t.Fatalf("expected index.existenceField to be nil")
}
}

2
pilosa.go
View file

@ -34,6 +34,8 @@ var (
ErrIndexExists = disco.ErrIndexExists
ErrIndexNotFound = errors.New("index not found")
ErrInvalidSchema = errors.New("invalid schema")
ErrForeignIndexNotFound = errors.New("foreign index not found")
// ErrFieldRequired is returned when no field is specified.

60
serializer.go Normal file
View file

@ -0,0 +1,60 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package pilosa
import (
"bytes"
"encoding/gob"
"fmt"
"github.com/pkg/errors"
)
// GobSerializer represents a Serializer that uses gob encoding. This is only
// used in tests; there's really no reason to use this instead of the proto
// serializer except that, as it's currently implemented, the proto serializer
// can't be used in internal tests (i.e test in the pilosa package) because the
// proto package imports the pilosa package, so it would result in circular
// imports. We really need all the pilosa types to be in a sub-package of
// pilosa, so that both proto and pilosa can import them without resulting in
// circular imports.
var GobSerializer Serializer = &gobSerializer{}
type gobSerializer struct{}
// Marshal is a gob-encoded implementation of the Serializer Marshal method.
func (s *gobSerializer) Marshal(msg Message) ([]byte, error) {
var buf bytes.Buffer
enc := gob.NewEncoder(&buf)
if err := enc.Encode(msg); err != nil {
return nil, errors.Wrap(err, "gob encoding message")
}
return buf.Bytes(), nil
}
// Unmarshal is a gob-encoded implementation of the Serializer Unmarshal method.
func (s *gobSerializer) Unmarshal(b []byte, m Message) error {
switch mt := m.(type) {
case *CreateIndexMessage, *CreateFieldMessage:
dec := gob.NewDecoder(bytes.NewReader(b))
err := dec.Decode(mt)
if err != nil {
return errors.Wrapf(err, "decoding %T", mt)
}
return nil
default:
panic(fmt.Sprintf("unhandled Message of type %T: %#v", mt, m))
}
}

62
test/field.go
View file

@ -15,72 +15,10 @@
package test
import (
"os"
"testing"
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/testhook"
)
// Field represents a test wrapper for pilosa.Field.
type Field struct {
*pilosa.Field
}
// newField returns a new instance of Field.
func newField(tb testing.TB, opts pilosa.FieldOption) *Field {
path, err := testhook.TempDir(tb, "pilosa-field-")
if err != nil {
panic(err)
}
// This path is probably wrong, but we don't care much because it's a scratch holder anyway.
field, err := pilosa.NewField(pilosa.NewHolder(path, nil), path, "i", "f", opts)
if err != nil {
panic(err)
}
return &Field{Field: field}
}
// mustOpenField returns a new, opened field at a temporary path. Panic on error.
func mustOpenField(tb testing.TB, opts pilosa.FieldOption) *Field {
f := newField(tb, opts)
if err := f.Open(); err != nil {
panic(err)
}
return f
}
// close closes the field and removes the underlying data.
func (f *Field) close() error { // nolint: unparam
defer os.RemoveAll(f.Path())
return f.Field.Close()
}
// reopen closes the index and reopens it.
func (f *Field) reopen() error {
if err := f.Field.Close(); err != nil {
return err
}
return f.Field.Open()
}
// Ensure field can set its cache
func TestField_SetCacheSize(t *testing.T) {
f := mustOpenField(t, pilosa.OptFieldTypeDefault())
defer f.close()
cacheSize := uint32(100)
// Set & retrieve field cache size.
if err := f.SetCacheSize(cacheSize); err != nil {
t.Fatal(err)
} else if q := f.CacheSize(); q != cacheSize {
t.Fatalf("unexpected field cache size: %d", q)
}
// Reload field and verify that it is persisted.
if err := f.reopen(); err != nil {
t.Fatal(err)
} else if q := f.CacheSize(); q != cacheSize {
t.Fatalf("unexpected field cache size (reopen): %d", q)
}
}

9
test/index.go
View file

@ -15,6 +15,7 @@
package test
import (
"context"
"testing"
"github.com/pilosa/pilosa/v2"
@ -32,7 +33,7 @@ func newIndex(tb testing.TB) *Index {
if err != nil {
panic(err)
}
h := pilosa.NewHolder(path, nil)
h := pilosa.NewHolder(path, pilosa.DefaultHolderConfig())
testhook.Cleanup(tb, func() {
h.Close()
})
@ -59,7 +60,11 @@ func (i *Index) Reopen() error {
if err := i.Index.Close(); err != nil {
return err
}
return i.Index.Open()
schema, err := i.Schemator.Schema(context.Background())
if err != nil {
return err
}
return i.OpenWithSchema(schema[i.Name()])
}
// CreateField creates a field with the given options.

34
view.go
View file

@ -575,26 +575,26 @@ func (v *view) rangeOp(qcx *Qcx, op pql.Token, bitDepth uint64, predicate int64)
return r, nil
}
// upgradeViewBSIv2 upgrades the fragments of v. Returns ok true if any fragment upgraded.
func upgradeViewBSIv2(v *view, bitDepth uint64) (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
func (v *view) bitDepth(shards []uint64) (uint64, error) {
var maxBitDepth uint64
if tmpPath, 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(tmpPath, frag.path()); err != nil {
return ok, errors.Wrap(err, "renaming after bsi v2 upgrade")
} else if err := frag.openStorage(true); err != nil {
return ok, errors.Wrap(err, "re-opening after bsi v2 upgrade")
for _, shard := range shards {
frag, ok := v.fragments[shard]
if !ok || frag == nil {
continue
}
bd, err := frag.bitDepth()
if err != nil {
return 0, errors.Wrapf(err, "getting fragment(%d) bit depth", shard)
}
if bd > maxBitDepth {
maxBitDepth = bd
}
}
return ok, nil
return maxBitDepth, nil
}
// ViewInfo represents schema information for a view.

2
view_internal_test.go
View file

@ -40,7 +40,7 @@ func mustOpenView(tb testing.TB, index, field, name string) *view {
cim := &CreateIndexMessage{
Index: index,
CreatedAt: 0,
Meta: &IndexOptions{},
Meta: IndexOptions{},
}
idx, err := h.createIndex(cim, false)