diff --git a/api.go b/api.go index 1cdbb004a..2933554de 100644 --- a/api.go +++ b/api.go @@ -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. diff --git a/broadcast.go b/broadcast.go index 141c43e36..cea51ed88 100644 --- a/broadcast.go +++ b/broadcast.go @@ -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. diff --git a/cluster.go b/cluster.go index 96b03be7c..465ecf840 100644 --- a/cluster.go +++ b/cluster.go @@ -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. diff --git a/dbshard.go b/dbshard.go index 3cd3bfceb..b2062abe2 100644 --- a/dbshard.go +++ b/dbshard.go @@ -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, } } diff --git a/dbshard_internal_test.go b/dbshard_internal_test.go index 00241d756..a5ae25348 100644 --- a/dbshard_internal_test.go +++ b/dbshard_internal_test.go @@ -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) diff --git a/encoding/proto/proto.go b/encoding/proto/proto.go index 465a44b63..121b632ca 100644 --- a/encoding/proto/proto.go +++ b/encoding/proto/proto.go @@ -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) { diff --git a/field.go b/field.go index 387456440..ef4e8c6ab 100644 --- a/field.go +++ b/field.go @@ -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 diff --git a/field_internal_test.go b/field_internal_test.go index d187d5ab9..ac20e89e3 100644 --- a/field_internal_test.go +++ b/field_internal_test.go @@ -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) + } +} diff --git a/fragment.go b/fragment.go index 3fb39c86c..b9c6eb154 100644 --- a/fragment.go +++ b/fragment.go @@ -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) diff --git a/fragment_internal_test.go b/fragment_internal_test.go index acb2cc915..2d096640f 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -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() diff --git a/holder.go b/holder.go index 24c6e939f..f813e1b99 100644 --- a/holder.go +++ b/holder.go @@ -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 diff --git a/index.go b/index.go index bd6d81c60..93de521ce 100644 --- a/index.go +++ b/index.go @@ -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) } diff --git a/index_internal_test.go b/index_internal_test.go index faed407c5..909278b2b 100644 --- a/index_internal_test.go +++ b/index_internal_test.go @@ -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") - } -} diff --git a/pilosa.go b/pilosa.go index f318a6eb4..370144c9d 100644 --- a/pilosa.go +++ b/pilosa.go @@ -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. diff --git a/serializer.go b/serializer.go new file mode 100644 index 000000000..15956739c --- /dev/null +++ b/serializer.go @@ -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)) + } +} diff --git a/test/field.go b/test/field.go index 817a72153..4663e4c0a 100644 --- a/test/field.go +++ b/test/field.go @@ -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) - } -} diff --git a/test/index.go b/test/index.go index 376ee6d65..d885d05da 100644 --- a/test/index.go +++ b/test/index.go @@ -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. diff --git a/view.go b/view.go index 4f0aa25ed..8e9f8e2b9 100644 --- a/view.go +++ b/view.go @@ -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. diff --git a/view_internal_test.go b/view_internal_test.go index b895122f4..336cc21b6 100644 --- a/view_internal_test.go +++ b/view_internal_test.go @@ -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)