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/field.go b/field.go index 3208c7c3d..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" @@ -591,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) @@ -709,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) @@ -730,32 +751,6 @@ func (f *Field) openViews() error { 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.options = applyDefaultOptions(opts) @@ -1291,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. @@ -1607,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. @@ -1814,31 +1797,6 @@ func applyDefaultOptions(o *FieldOptions) FieldOptions { 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, - } -} - // MarshalJSON marshals FieldOptions to JSON such that // only those attributes associated to the field type // are included. diff --git a/field_internal_test.go b/field_internal_test.go index 9f2fd8134..ac20e89e3 100644 --- a/field_internal_test.go +++ b/field_internal_test.go @@ -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 c3e1e862d..91c77b2dc 100644 --- a/fragment.go +++ b/fragment.go @@ -238,6 +238,27 @@ 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) { + var maxBitDepth uint64 + + 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 { + if maxRowID+1-bsiOffsetBit > maxBitDepth { + maxBitDepth = uint64(maxRowID + 1 - bsiOffsetBit) + } + return maxBitDepth, nil +} + type FragmentInfo struct { BitmapInfo roaring.BitmapInfo BlockChecksums []FragmentBlock `json:"BlockChecksums,omitempty"` diff --git a/index.go b/index.go index f0b74e733..93de521ce 100644 --- a/index.go +++ b/index.go @@ -227,6 +227,19 @@ func (i *Index) open(idx *disco.Index) (err error) { 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") @@ -411,6 +424,26 @@ func (i *Index) openExistenceField() error { return nil } +// 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() + } + return nil +} + // Close closes the index and its fields. func (i *Index) Close() error { i.mu.Lock() @@ -726,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 diff --git a/view.go b/view.go index 9d82fc4b4..8e9f8e2b9 100644 --- a/view.go +++ b/view.go @@ -575,6 +575,28 @@ func (v *view) rangeOp(qcx *Qcx, op pql.Token, bitDepth uint64, predicate int64) return r, nil } +func (v *view) bitDepth(shards []uint64) (uint64, error) { + var maxBitDepth uint64 + + 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 maxBitDepth, nil +} + // ViewInfo represents schema information for a view. type ViewInfo struct { Name string `json:"name"`