From 035073555a06915e5ce693f9e51dd610bb7ae6bd Mon Sep 17 00:00:00 2001 From: "Jason E. Aten" Date: Thu, 17 Dec 2020 21:25:55 +0000 Subject: [PATCH] pilosa: only open views with data - Previously, on timequantum schemas, we would create and open a view for the cartesian product of every possible view and shard. - This caused us to be very slow on re-open, and to use lots of memory for views that held nothing. - This change makes startup faster, memory use much lower, and should speed migration. --- bluegreentx.go | 11 ++ bolt.go | 21 +++- bolt_test.go | 28 +++-- catcher.go | 6 + dbshard.go | 223 +++++++++++++++++++++++++++++++++++--- dbshard_internal_test.go | 170 ++++++++++++++++++++++++++--- field.go | 94 +++++----------- fragment_internal_test.go | 2 +- holder.go | 2 +- index.go | 15 ++- rbf.go | 5 + rbf/ingest_test.go | 1 + rbf/tx.go | 16 +++ rrtx.go | 172 ++++++++++++++++++++++++++++- short_txkey/txkey.go | 16 +++ stattx.go | 6 + tx.go | 9 +- txfactory.go | 7 +- txkey/txkey.go | 16 +++ view.go | 133 ++++++++++++----------- view_internal_test.go | 2 +- 21 files changed, 769 insertions(+), 186 deletions(-) diff --git a/bluegreentx.go b/bluegreentx.go index 738d146c8..62f1c0bd8 100644 --- a/bluegreentx.go +++ b/bluegreentx.go @@ -22,6 +22,8 @@ import ( "sync" "github.com/pilosa/pilosa/v2/roaring" + txkey "github.com/pilosa/pilosa/v2/short_txkey" + //txkey "github.com/pilosa/pilosa/v2/txkey" ) // blueGreenTx runs two Tx together and notices differences in their output. @@ -194,6 +196,15 @@ func (c *blueGreenTx) Readonly() bool { return b } +// for now we just return B's list, since this is involved in +// holder Open which can happen before any blue-green is done; +// in fact this is instrumental in setting up the sync from +// green to blue. +func (c *blueGreenTx) GetSortedFieldViewList(idx *Index, shard uint64) (fvB []txkey.FieldView, errB error) { + fvB, errB = c.b.GetSortedFieldViewList(idx, shard) + return +} + func (c *blueGreenTx) NewTxIterator(index, field, view string, shard uint64) *roaring.Iterator { c.checker.see(index, field, view, shard) // can't really do simultaneous iteration on A and B, so punt and diff --git a/bolt.go b/bolt.go index 623a1001a..b8a3f25be 100644 --- a/bolt.go +++ b/bolt.go @@ -32,7 +32,12 @@ import ( "github.com/pilosa/pilosa/v2/rbf" rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg" "github.com/pilosa/pilosa/v2/roaring" - "github.com/pilosa/pilosa/v2/txkey" + + // On Bolt only, we still use the long txkey, because + // this allows Max() to work readily. + // + "github.com/pilosa/pilosa/v2/short_txkey" + txkey "github.com/pilosa/pilosa/v2/txkey" "github.com/pkg/errors" bolt "go.etcd.io/bbolt" ) @@ -632,6 +637,7 @@ func (tx *BoltTx) Remove(index, field, view string, shard uint64, a ...uint64) ( } func (tx *BoltTx) addOrRemove(index, field, view string, shard uint64, batched, remove bool, a ...uint64) (changeCount int, err error) { + // pure hack to match RoaringTx defer func() { if !remove && !batched { @@ -1454,6 +1460,19 @@ func (tx *BoltTx) Dump(short bool, shard uint64) { fmt.Printf("%v\n", stringifiedBoltKeysTx(tx, short)) } +func (tx *BoltTx) GetSortedFieldViewList(idx *Index, shard uint64) (fvs []short_txkey.FieldView, err error) { + bkt := tx.tx.Bucket(bucketCT) + err = bkt.ForEach(func(bkey, v []byte) error { + fv := txkey.FieldViewFromFullKey(bkey) + var shortFV short_txkey.FieldView + shortFV.Field = fv.Field + shortFV.View = fv.View + fvs = append(fvs, shortFV) + return nil + }) + return +} + // stringifiedBoltKeysTx reports all the bolt keys and a // corresponding blake3 hash viewable by txn within the entire // bolt database. diff --git a/bolt_test.go b/bolt_test.go index 9a6961d29..3c41ddc41 100644 --- a/bolt_test.go +++ b/bolt_test.go @@ -115,16 +115,14 @@ func TestBolt_DeleteFragment(t *testing.T) { dbwrap, clean := mustOpenEmptyBoltWrapper("TestBolt_DeleteFragment") defer clean() defer dbwrap.Close() - index, field, view, shard0 := "i", "f", "v", uint64(0) + index, field, shard := "i", "f", uint64(0) tx, _ := dbwrap.NewTx(writable, index, Txo{}) - shard1 := uint64(1) - bits := []uint64{0, 3, 1 << 16, 1<<16 + 3, 8 << 16} - shards := []uint64{shard0, shard1} - for _, s := range shards { + views := []string{"v1", "v2"} + for _, view := range views { for _, v := range bits { - changed, err := tx.Add(index, field, view, s, doBatched, v) + changed, err := tx.Add(index, field, view, shard, doBatched, v) if changed <= 0 { panic("should have changed") } @@ -132,9 +130,9 @@ func TestBolt_DeleteFragment(t *testing.T) { } } - for _, s := range shards { + for _, view := range views { for _, v := range bits { - exists, err := tx.Contains(index, field, view, s, v) + exists, err := tx.Contains(index, field, view, shard, v) panicOn(err) if !exists { panic("ARG bitvalue was NOT SET!!!") @@ -146,23 +144,23 @@ func TestBolt_DeleteFragment(t *testing.T) { // end of setup - survivor := shard0 - victim := shard1 - err = dbwrap.DeleteFragment(index, field, view, victim, nil) + victim := "v1" + survivor := "v2" + err = dbwrap.DeleteFragment(index, field, victim, shard, nil) panicOn(err) tx, _ = dbwrap.NewTx(!writable, index, Txo{}) defer tx.Rollback() - for _, s := range shards { + for _, view := range views { for _, v := range bits { - exists, err := tx.Contains(index, field, view, s, v) + exists, err := tx.Contains(index, field, view, shard, v) panicOn(err) - if s == survivor { + if view == survivor { if !exists { panic(fmt.Sprintf("ARG survivor died : bit %v", v)) } - } else if s == victim { // victim, should have been deleted + } else if view == victim { // victim, should have been deleted if exists { panic(fmt.Sprintf("ARG victim lived : bit %v", v)) } diff --git a/catcher.go b/catcher.go index 960e8527b..671ee7662 100644 --- a/catcher.go +++ b/catcher.go @@ -19,6 +19,8 @@ import ( "io" "github.com/pilosa/pilosa/v2/roaring" + txkey "github.com/pilosa/pilosa/v2/short_txkey" + //txkey "github.com/pilosa/pilosa/v2/txkey" ) // catcher is useful to report error locations with a @@ -311,3 +313,7 @@ func (c *catcherTx) Sn() int64 { func (c *catcherTx) ApplyFilter(index, field, view string, shard uint64, ckey uint64, filter roaring.BitmapFilter) (err error) { return GenericApplyFilter(c, index, field, view, shard, ckey, filter) } + +func (c *catcherTx) GetSortedFieldViewList(idx *Index, shard uint64) (fvs []txkey.FieldView, err error) { + return c.b.GetSortedFieldViewList(idx, shard) +} diff --git a/dbshard.go b/dbshard.go index 2c9e2989e..cdffdc606 100644 --- a/dbshard.go +++ b/dbshard.go @@ -24,6 +24,8 @@ import ( "sync" rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg" + txkey "github.com/pilosa/pilosa/v2/short_txkey" + //txkey "github.com/pilosa/pilosa/v2/txkey" "github.com/pkg/errors" ) @@ -259,6 +261,36 @@ type shardSet struct { readonlyVer int64 } +func (a *shardSet) unionInPlace(b *shardSet) { + shards := b.CloneMaybe() + for shard := range shards { + a.add(shard) + } +} + +func (a *shardSet) equals(b *shardSet) bool { + if len(a.shards) != len(b.shards) { + return false + } + for shardInA := range a.shards { + _, ok := b.shards[shardInA] + if !ok { + return false + } + } + return true + +} + +func (ss *shardSet) String() (r string) { + r = "[" + for k := range ss.shards { + r += fmt.Sprintf("%v, ", k) + } + r += "]" + return +} + func (ss *shardSet) add(shard uint64) { _, already := ss.shards[shard] if !already { @@ -295,6 +327,12 @@ func newShardSet() *shardSet { shards: make(map[uint64]bool), } } +func newShardSetFromMap(m map[uint64]bool) *shardSet { + return &shardSet{ + shards: m, + shardsVer: 1, + } +} // HasData returns true if the database has at least one key. // For roaring it returns true if we a fragment stored. @@ -556,9 +594,12 @@ func (per *DBPerShard) updateIndex2ShardCacheWithNewShard(dbs *DBShard) { } func (per *DBPerShard) GetDBShard(index string, shard uint64, idx *Index) (dbs *DBShard, err error) { - per.Mu.Lock() defer per.Mu.Unlock() + return per.unprotectedGetDBShard(index, shard, idx) +} + +func (per *DBPerShard) unprotectedGetDBShard(index string, shard uint64, idx *Index) (dbs *DBShard, err error) { dbi, ok := per.dbh.Index[index] if !ok { @@ -689,6 +730,11 @@ func (f *TxFactory) GetShardsForIndex(idx *Index, roaringViewPath string, requir // // when a new DBShard is made, we will update the list of shards then. Thus // the per.index2shard should always be up to date AFTER the first call here. +// +// Note: we cannot here call GetView2ShardsMapForIndex() because that only ever +// returns the green data and we are used during migration for both blue +// and green. +// func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, roaringViewPath string, requireData bool) (shardMap map[uint64]bool, err error) { // use the cache, always @@ -696,17 +742,10 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r defer per.Mu.Unlock() if ty == roaringTxn && roaringViewPath != "" { - rx := &RoaringTx{ - Index: idx, - } - sos, err := rx.SliceOfShards("", "", "", roaringViewPath) + shardMap, err := roaringMapOfShards(roaringViewPath) if err != nil { return nil, err } - shardMap = make(map[uint64]bool) - for _, shard := range sos { - shardMap[shard] = true - } return shardMap, nil } @@ -733,19 +772,16 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r if ty == roaringTxn { // INVAR: roaringViewPath == "", because the other case is // handled above. - rx := &RoaringTx{ - Index: idx, - } fields := idx.Fields() for _, field := range fields { for _, view := range field.views() { - sos, err := rx.SliceOfShards("", "", "", view.path) + shardMap, err := roaringMapOfShards(view.path) if err != nil { return nil, errors.Wrap(err, fmt.Sprintf( "TypedDBPerShardGetLocalShardsForIndex roaringTxn view.path='%v'", view.path)) } - for _, shard := range sos { + for shard := range shardMap { setOfShards.add(shard) } } @@ -788,7 +824,7 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r hasData := false if requireData { - hasData, err = per.TypedIndexShardHasData(ty, idx, shard) + hasData, err = per.unprotectedTypedIndexShardHasData(ty, idx, shard) if err != nil { return nil, err } @@ -803,7 +839,7 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r return setOfShards.CloneMaybe(), nil } -func (per *DBPerShard) TypedIndexShardHasData(ty txtype, idx *Index, shard uint64) (hasData bool, err error) { +func (per *DBPerShard) unprotectedTypedIndexShardHasData(ty txtype, idx *Index, shard uint64) (hasData bool, err error) { whichty := 0 if len(per.types) == 2 { if ty == per.types[1] { @@ -815,7 +851,7 @@ func (per *DBPerShard) TypedIndexShardHasData(ty txtype, idx *Index, shard uint6 } // make the dbs if it doesn't get exist - dbs, err := per.GetDBShard(idx.name, shard, idx) + dbs, err := per.unprotectedGetDBShard(idx.name, shard, idx) if err != nil { return false, errors.Wrap(err, fmt.Sprintf("DBPerShard.TypedIndexShardHasData() "+ "per.GetDBShard(index='%v', shard='%v', ty='%v')", idx.name, shard, ty.String())) @@ -1043,3 +1079,156 @@ func (dbs *DBShard) verifyBlueEqualsGreen() (err error) { return nil } + +type FieldView2Shards struct { + // field -> view -> *shardSet + m map[string]map[string]*shardSet +} + +func (vs *FieldView2Shards) getViewsForField(field string) map[string]*shardSet { + return vs.m[field] +} + +func (vs *FieldView2Shards) has(field, view string, shard uint64) bool { + vw, ok := vs.m[field] + if !ok { + return false + } + ss, ok := vw[view] + if !ok { + return false + } + shardMap := ss.CloneMaybe() + return shardMap[shard] +} + +func (vs *FieldView2Shards) addViewShardSet(fv txkey.FieldView, ss *shardSet) { + + f, ok := vs.m[fv.Field] + if !ok { + f = make(map[string]*shardSet) + vs.m[fv.Field] = f + } + // INVAR: f is ready to take ss. + + // existing stuff to merge with? + prior, ok := f[fv.View] + if !ok { + f[fv.View] = ss + return + } + // merge ss and prior. No need to put the union back into f[fv.View] + // because prior is a pointer. + prior.unionInPlace(ss) +} + +func (a *FieldView2Shards) equals(b *FieldView2Shards) bool { + if a == nil && b == nil { + return true + } + if a == nil || b == nil { + return false + } + if len(a.m) != len(b.m) { + return false + } + for field, viewmapA := range a.m { + viewmapB, ok := b.m[field] + if !ok { + return false + } + if len(viewmapB) != len(viewmapA) { + return false + } + for k, va := range viewmapA { + vb, ok := viewmapB[k] + if !ok { + return false + } + if !va.equals(vb) { + return false + } + } + } + return true +} + +func NewFieldView2Shards() *FieldView2Shards { + return &FieldView2Shards{ + m: make(map[string]map[string]*shardSet), // expected response from GetView2ShardMapForIndex + } +} + +func (vs *FieldView2Shards) addShard(fv txkey.FieldView, shard uint64) { + viewmap, ok := vs.m[fv.Field] + if !ok { + viewmap = make(map[string]*shardSet) + vs.m[fv.Field] = viewmap + } + ss, ok := viewmap[fv.View] + if !ok { + ss = newShardSet() + viewmap[fv.View] = ss + } + ss.add(shard) +} + +func (vs *FieldView2Shards) String() (r string) { + r = "\n" + for field, viewmap := range vs.m { + for view, shards := range viewmap { + r += fmt.Sprintf("field '%v' view:'%v' shards:%v\n", field, view, shards) + } + } + r += "\n" + return +} + +// Note: cannot call this during migration, because +// it only ever returns the green shards if we are in blue-green. +func (per *DBPerShard) GetFieldView2ShardsMapForIndex(idx *Index) (vs *FieldView2Shards, err error) { + + // for blue-green, it does matter that we return green, so we can migrate from it. + ty := per.types[0] + if per.isBlueGreen { + ty = per.types[1] + } + + switch ty { + case roaringTxn: + return roaringGetFieldView2Shards(idx) + default: + vs = NewFieldView2Shards() + + shardMap, err := per.TypedDBPerShardGetShardsForIndex(ty, idx, "", true) + if err != nil { + return nil, err + } + + for shard := range shardMap { + dbs, err := per.GetDBShard(idx.name, shard, idx) + if err != nil { + return nil, errors.Wrap(err, "DBPerShard.GetFieldView2ShardsMapForIndex GetDBShard()") + } + fieldviews, err := dbs.AllFieldViews() + if err != nil { + return nil, errors.Wrap(err, "DBPerShard.GetFieldView2ShardsMapForIndex dbs.AllFieldViews()") + } + for _, fv := range fieldviews { + vs.addShard(fv, shard) + } + } + } + + return +} + +func (dbs *DBShard) AllFieldViews() (fvs []txkey.FieldView, err error) { + + tx, err := dbs.NewTx(!writable, dbs.idx.name, Txo{Write: !writable, Shard: dbs.Shard, Index: dbs.idx, dbs: dbs}) + if err != nil { + return nil, errors.Wrap(err, fmt.Sprintf("dbshard.NewTx for index '%v', shard %v", dbs.idx.name, dbs.Shard)) + } + defer tx.Rollback() + return tx.GetSortedFieldViewList(dbs.idx, dbs.Shard) +} diff --git a/dbshard_internal_test.go b/dbshard_internal_test.go index 02340035c..18858485c 100644 --- a/dbshard_internal_test.go +++ b/dbshard_internal_test.go @@ -23,6 +23,9 @@ import ( "testing" "github.com/pilosa/pilosa/v2/rbf" + "github.com/pilosa/pilosa/v2/shardwidth" + txkey "github.com/pilosa/pilosa/v2/short_txkey" + //txkey "github.com/pilosa/pilosa/v2/txkey" ) // Shard per db evaluation @@ -68,12 +71,22 @@ func TestShardPerDB_SetBit(t *testing.T) { // test that we find all *local* shards func Test_DBPerShard_GetShardsForIndex_LocalOnly(t *testing.T) { + tmpdir, err := ioutil.TempDir("", "Test_DBPerShard_GetShardsForIndex_LocalOnly") panicOn(err) orig := os.Getenv("PILOSA_TXSRC") defer os.Setenv("PILOSA_TXSRC", orig) // must restore or will mess up other tests! + v2s := NewFieldView2Shards() + stdShardSet := newShardSet() + for _, shard := range []uint64{93, 223, 221, 215, 219, 217} { + stdShardSet.add(shard) + } + for _, field := range []string{"f", "_exists"} { + v2s.addViewShardSet(txkey.FieldView{Field: field, View: "standard"}, stdShardSet) + } + for _, src := range []string{"roaring", "bolt", "rbf"} { os.Setenv("PILOSA_TXSRC", src) @@ -81,10 +94,12 @@ func Test_DBPerShard_GetShardsForIndex_LocalOnly(t *testing.T) { // must make Holder AFTER setting src. holder := NewHolder(tmpdir, nil) - makeSampleRoaringDir(tmpdir, src, 1, holder) - - idx, err := NewIndex(holder, tmpdir, "rick") - panicOn(err) + index := "rick" + idx := makeSampleRoaringDir(tmpdir, index, src, 1, holder, v2s) + if idx == nil { + idx, err = NewIndex(holder, filepath.Join(tmpdir, index), index) + panicOn(err) + } estd := "rick/_exists/views/standard" std := "rick/f/views/standard" @@ -105,6 +120,56 @@ func Test_DBPerShard_GetShardsForIndex_LocalOnly(t *testing.T) { panic(fmt.Sprintf("missing shard=%v from shards='%#v'", shard, shards)) } } + + // check GetSortedFieldViewList() and roaringGetFieldView2Shards() + vs, err := roaringGetFieldView2Shards(idx) + panicOn(err) + + for _, shard := range []uint64{93, 223, 221, 215, 219, 217} { + tx := idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Shard: shard}) + fvs, err := tx.GetSortedFieldViewList(idx, shard) + panicOn(err) + // expect these same two field/views for all 6 shards + expect0 := txkey.FieldView{Field: "_exists", View: "standard"} + expect1 := txkey.FieldView{Field: "f", View: "standard"} + if len(fvs) != 2 { + panic(fmt.Sprintf("fvs should be len 2, got '%#v'", fvs)) + } + if fvs[0] != expect0 { + panic(fmt.Sprintf("expected fvs[0]='%#v', but got '%#v'", expect0, fvs[0])) + } + if fvs[1] != expect1 { + panic(fmt.Sprintf("expected fvs[1]='%#v', but got '%#v'", expect1, fvs[1])) + } + + for _, fv := range fvs { + if !vs.has(fv.Field, fv.View, shard) { + panic(fmt.Sprintf("vs did not contain fv='%#v' for shard %v", fv, shard)) + } + } + tx.Rollback() + } + } else { + // non-roaring: rbf, bolt + + for _, shard := range []uint64{93, 223, 221, 215, 219, 217} { + tx := idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Shard: shard}) + fvs, err := tx.GetSortedFieldViewList(idx, shard) + panicOn(err) + // expect these same two field/views for all 6 shards + expect0 := txkey.FieldView{Field: "_exists", View: "standard"} + expect1 := txkey.FieldView{Field: "f", View: "standard"} + if len(fvs) != 2 { + panic(fmt.Sprintf("fvs should be len 2, got '%#v'", fvs)) + } + if fvs[0] != expect0 { + panic(fmt.Sprintf("expected fvs[0]='%#v', but got '%#v'", expect0, fvs[0])) + } + if fvs[1] != expect1 { + panic(fmt.Sprintf("expected fvs[1]='%#v', but got '%#v'", expect1, fvs[1])) + } + tx.Rollback() + } } holder.Close() } @@ -150,11 +215,12 @@ rick.index.txstores@@@/store-rbfdb@@/shard.0223-rbfdb@ `, } -func makeSampleRoaringDir(root, txsrc string, minBytes int, h *Holder) { +func makeSampleRoaringDir(root, index, txsrc string, minBytes int, h *Holder, view2shards *FieldView2Shards) (idx *Index) { - index := "rick" shards := []uint64{0, 93, 215, 217, 219, 221, 223} fns := strings.Split(sampleRoaringDirList[txsrc], "\n") + firstDone := false + for i, fn := range fns { if fn == "" { continue @@ -165,13 +231,15 @@ func makeSampleRoaringDir(root, txsrc string, minBytes int, h *Holder) { shard = shards[i] } switch txsrc { - case "bolt": - makeBolttestDB(root+sep+fn, h, shard) - helperCreateDBShard(h, index, shard) - continue - case "rbf": - makeRBFtestDB(root+sep+fn, h, shard) - helperCreateDBShard(h, index, shard) + case "bolt", "rbf": + idx = helperCreateDBShard(h, index, shard) + + // first time only, we'll actually make all the shards at this point because + // view2shards has them all anyway. + if !firstDone { + firstDone = true + makeTxTestDBWithViewsShards(h, idx, view2shards) + } continue } @@ -185,16 +253,22 @@ func makeSampleRoaringDir(root, txsrc string, minBytes int, h *Holder) { } fd.Close() } + return } -func helperCreateDBShard(h *Holder, index string, shard uint64) { +func helperCreateDBShard(h *Holder, index string, shard uint64) *Index { idx, err := h.CreateIndexIfNotExists(index, IndexOptions{}) panicOn(err) dbs, err := h.txf.dbPerShard.GetDBShard(index, shard, idx) panicOn(err) _ = dbs + return idx } +// keep the ocd linter happy +var _ = makeBolttestDB +var _ = makeRBFtestDB + func makeBolttestDB(path string, h *Holder, shard uint64) { i := uint64(1) w, _ := mustOpenEmptyBoltWrapper(path) @@ -222,3 +296,71 @@ func makeRBFtestDB(path string, h *Holder, shard uint64) { err = tx.Commit() panicOn(err) } + +func makeTxTestDBWithViewsShards(holder *Holder, idx *Index, exp *FieldView2Shards) { + + // TODO(jea): need date time quantum views!! + batched := false + for field, viewmap := range exp.m { + for view, shset := range viewmap { + + ss := shset.CloneMaybe() + for shard := range ss { + + // simply write 1 bit to each shard to force its creation. + bits := []uint64{(shard << shardwidth.Exponent) + 1} + tx := idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard}) + changeCount, err := tx.Add(idx.name, field, view, shard, batched, bits...) + panicOn(err) + if changeCount != len(bits) { + panic(fmt.Sprintf("writing field '%v', view '%v' shard '%v', expected changeCount to equal len bits = %v but was %v", field, view, shard, len(bits), changeCount)) + } + + panicOn(tx.Commit()) + } + } + } + +} + +// test that rbf can give us a map[view]*shardSet +func Test_DBPerShard_GetFieldView2Shards_map_from_RBF(t *testing.T) { + tmpdir, err := ioutil.TempDir("", "Test_DBPerShard_GetFieldView2Shards_map_from_RBF") + panicOn(err) + + orig := os.Getenv("PILOSA_TXSRC") + defer os.Setenv("PILOSA_TXSRC", orig) // must restore or will mess up other tests! + + os.Setenv("PILOSA_TXSRC", "rbf") + + // must make Holder AFTER setting src. + holder := NewHolder(tmpdir, nil) + defer holder.Close() + + index := "rick" + field := "f" + idx, err := holder.createIndex(index, IndexOptions{}) + panicOn(err) + + exp := NewFieldView2Shards() + + stdShardSet := newShardSet() + stdShardSet.add(12) + stdShardSet.add(15) + exp.addViewShardSet(txkey.FieldView{Field: field, View: "standard"}, stdShardSet) + + hrShardSet := newShardSet() + hrShardSet.add(7) + exp.addViewShardSet(txkey.FieldView{Field: field, View: "standard_2019092416"}, hrShardSet) + + makeTxTestDBWithViewsShards(holder, idx, exp) + + // setup is done + view2shard, err := holder.txf.GetFieldView2ShardsMapForIndex(idx) + panicOn(err) + + // compare against setup + if !view2shard.equals(exp) { + panic(fmt.Sprintf("expected '%v' but got view2shard '%v'", exp, view2shard)) + } +} diff --git a/field.go b/field.go index 970f4b6f6..baec3a931 100644 --- a/field.go +++ b/field.go @@ -38,7 +38,6 @@ import ( "github.com/pilosa/pilosa/v2/testhook" "github.com/pilosa/pilosa/v2/tracing" "github.com/pkg/errors" - "golang.org/x/sync/errgroup" ) // Default field settings. @@ -581,6 +580,7 @@ func (f *Field) Open() error { } 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") } @@ -737,82 +737,44 @@ func (f *Field) ForeignIndex() string { return f.options.ForeignIndex } -var fieldQueue = make(chan struct{}, 16) - // openViews opens and initializes the views inside the field. func (f *Field) openViews() error { - file, err := os.Open(filepath.Join(f.path, "views")) - if os.IsNotExist(err) { + view2shards := f.idx.fieldView2shard.getViewsForField(f.name) + if view2shards == nil { + // no data return nil - } else if err != nil { - return errors.Wrap(err, "opening view directory") } - defer file.Close() - fis, err := file.Readdir(0) - if err != nil { - return errors.Wrap(err, "reading directory") - } - eg, ctx := errgroup.WithContext(context.Background()) - var mu sync.Mutex + for name, shardset := range view2shards { -fileLoop: - for _, loopFi := range fis { - select { - case <-ctx.Done(): - break fileLoop - default: - fi := loopFi - if !fi.IsDir() { - continue - } - // Skip embedded db files too. - if f.holder.txf.IsTxDatabasePath(fi.Name()) { - continue - } + 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) + } - fieldQueue <- struct{}{} - eg.Go(func() error { - defer func() { - <-fieldQueue - }() - name := filepath.Base(fi.Name()) - f.holder.Logger.Debugf("open index/field/view: %s/%s/%s", f.index, f.name, fi.Name()) - - view := f.newView(f.viewPath(name), name) - if err := view.open(); 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.open(); err != nil { - return fmt.Errorf("re-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) - mu.Lock() - f.viewMap[view.name] = view - mu.Unlock() - return nil - }) + } } - } - return eg.Wait() + 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 + } + return nil } // loadMeta reads meta data for the field, if any. @@ -1209,7 +1171,7 @@ func (f *Field) createViewIfNotExistsBase(name string) (*view, bool, error) { } view := f.newView(f.viewPath(name), name) - if err := view.open(); err != nil { + if err := view.openEmpty(); err != nil { return nil, false, errors.Wrap(err, "opening view") } view.rowAttrStore = f.rowAttrStore diff --git a/fragment_internal_test.go b/fragment_internal_test.go index a120d0b5c..fa45aace4 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -5788,7 +5788,7 @@ func BenchmarkImportMutexSampleData(b *testing.B) { func testOneParallelSlice(t *testing.T, p *parallelSlices) { // the easy answer seen := make(map[uint64]uint64, len(p.cols)) - t.Logf("cols %d, rows %d", p.cols, p.rows) + //t.Logf("cols %d, rows %d", p.cols, p.rows) for i, c := range p.cols { seen[c] = p.rows[i] } diff --git a/holder.go b/holder.go index c3e6ff01a..256871058 100644 --- a/holder.go +++ b/holder.go @@ -55,7 +55,7 @@ const ( func init() { // needed to get the most I/O throughtpu. - runtime.GOMAXPROCS(128) + runtime.GOMAXPROCS(runtime.NumCPU()) // For performance tuning, leave these readily available: // CPUProfileForDur(time.Minute, "server.cpu.pprof") diff --git a/index.go b/index.go index 9cbe19fed..6129289f5 100644 --- a/index.go +++ b/index.go @@ -71,6 +71,9 @@ type Index struct { // Instantiates new translation stores OpenTranslateStore OpenTranslateStoreFunc + + // track the subset of shards available to our views + fieldView2shard *FieldView2Shards } // NewIndex returns an existing (but possibly empty) instance of @@ -190,6 +193,16 @@ func (i *Index) open(withTimestamp bool) (err error) { 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 + // of what fields/views/shards are present with data up front. + fieldView2shard, err := i.holder.txf.GetFieldView2ShardsMapForIndex(i) + if err != nil { + return errors.Wrap(err, fmt.Sprintf("i.holder.txf.GetFieldView2ShardsMapForIndex('%v')", i.name)) + } + i.fieldView2shard = fieldView2shard + i.holder.Logger.Debugf("open fields for index: %s", i.name) if err := i.openFields(withTimestamp); err != nil { return errors.Wrap(err, "opening fields") @@ -294,7 +307,7 @@ fileLoop: // up a foreign index. fld.holder = i.holder - // open all the views + // open the views we have data for. if err := fld.Open(); err != nil { return fmt.Errorf("open field: name=%s, err=%s", fld.Name(), err) } diff --git a/rbf.go b/rbf.go index 28b7932f8..dff5a1123 100644 --- a/rbf.go +++ b/rbf.go @@ -30,6 +30,7 @@ import ( rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg" "github.com/pilosa/pilosa/v2/roaring" txkey "github.com/pilosa/pilosa/v2/short_txkey" + //txkey "github.com/pilosa/pilosa/v2/txkey" "github.com/pkg/errors" ) @@ -430,6 +431,10 @@ func (tx *RBFTx) ApplyFilter(index, field, view string, shard uint64, ckey uint6 return tx.tx.ApplyFilter(rbfName(index, field, view, shard), ckey, filter) } +func (tx *RBFTx) GetSortedFieldViewList(idx *Index, shard uint64) (fvs []txkey.FieldView, err error) { + return tx.tx.GetSortedFieldViewList() +} + // rbfName returns a NULL-separated key used for identifying bitmap maps in RBF. func rbfName(index, field, view string, shard uint64) string { return string(txkey.Prefix(index, field, view, shard)) diff --git a/rbf/ingest_test.go b/rbf/ingest_test.go index 3ba99e84d..6dcfdc9e6 100644 --- a/rbf/ingest_test.go +++ b/rbf/ingest_test.go @@ -27,6 +27,7 @@ import ( "github.com/pilosa/pilosa/v2/rbf/cfg" "github.com/pilosa/pilosa/v2/roaring" + // "github.com/pilosa/pilosa/v2/txkey" txkey "github.com/pilosa/pilosa/v2/short_txkey" ) diff --git a/rbf/tx.go b/rbf/tx.go index dff63bd1b..705edfc5b 100644 --- a/rbf/tx.go +++ b/rbf/tx.go @@ -26,6 +26,7 @@ import ( "github.com/pilosa/pilosa/v2/hash" "github.com/pilosa/pilosa/v2/roaring" txkey "github.com/pilosa/pilosa/v2/short_txkey" + //txkey "github.com/pilosa/pilosa/v2/txkey" ) var _ = txkey.ToString @@ -2032,6 +2033,21 @@ func (tx *Tx) PageData(pgno uint32) ([]byte, error) { return buf, err } +func (tx *Tx) GetSortedFieldViewList() (fvs []txkey.FieldView, _ error) { + records, err := tx.RootRecords() + if err != nil { + return nil, err + } + it := records.Iterator() + for !it.Done() { + k, _ := it.Next() + root := k.(string) + fv := txkey.FieldViewFromPrefix([]byte(root)) + fvs = append(fvs, fv) + } + return +} + type PageInfo interface { pageInfo() } diff --git a/rrtx.go b/rrtx.go index bd16f04d8..0cfb5fba9 100644 --- a/rrtx.go +++ b/rrtx.go @@ -20,6 +20,7 @@ import ( "io" "os" "path/filepath" + "sort" "strconv" "strings" "sync" @@ -28,6 +29,8 @@ import ( "github.com/pilosa/pilosa/v2/rbf" rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg" "github.com/pilosa/pilosa/v2/roaring" + txkey "github.com/pilosa/pilosa/v2/short_txkey" + //txkey "github.com/pilosa/pilosa/v2/txkey" "github.com/pkg/errors" ) @@ -67,9 +70,10 @@ func (tx *RoaringTx) UseRowCache() bool { return rbf.EnableRowCache() } -func (tx *RoaringTx) SliceOfShards(index, field, view, optionalViewPath string) (sliceOfShards []uint64, err error) { +// based on view.openFragments() +func roaringMapOfShards(optionalViewPath string) (shardMap map[uint64]bool, err error) { - // SliceOfShards is based on view.openFragments() + shardMap = make(map[uint64]bool) path := filepath.Join(optionalViewPath, "fragments") file, err := os.Open(path) @@ -103,7 +107,7 @@ func (tx *RoaringTx) SliceOfShards(index, field, view, optionalViewPath string) //tx.Index.holder.Logger.Debugf("WARNING: couldn't use non-integer file as shard in index/field/view %s/%s/%s: %s", index, field, view, fi.Name()) continue } - sliceOfShards = append(sliceOfShards, shard) + shardMap[shard] = true } return } @@ -419,6 +423,168 @@ func (tx *RoaringTx) Sn() int64 { return tx.sn } +func roaringGetFieldView2Shards(idx *Index) (vs *FieldView2Shards, err error) { + vs = NewFieldView2Shards() + + // A) open the index directory + f, err := os.Open(idx.path) + if err != nil { + return nil, errors.Wrap(err, "opening directory") + } + defer f.Close() + + fieldFIs, err := f.Readdir(0) + if err != nil { + return nil, errors.Wrap(err, "reading directory") + } + + //vv("roaringGetFieldView2Shards A) opened index path '%v'", idx.path) + + // B) read the name of each field under the index + for _, loopFieldFi := range fieldFIs { + fieldFI := loopFieldFi + if !fieldFI.IsDir() { + continue + } + field := fieldFI.Name() + + //vv("roaringGetFieldView2Shards B) on field '%v'", field) + + fieldPath := filepath.Join(idx.path, field) + + // Skip embedded db files too. + if idx.holder.txf.IsTxDatabasePath(field) { + continue + } + viewsDir := filepath.Join(fieldPath, "views") + file, err := os.Open(viewsDir) + if os.IsNotExist(err) { + //return nil + continue + } else if err != nil { + return nil, errors.Wrapf(err, "opening view directory '%v'", viewsDir) + } + defer file.Close() + + // C) read the name of each view under the field + + viewFIs, err := file.Readdir(0) + if err != nil { + return nil, errors.Wrapf(err, "reading views directory '%v'", viewsDir) + } + for _, viewFI := range viewFIs { + + if !viewFI.IsDir() { + continue + } + view := viewFI.Name() + roaringViewPath := filepath.Join(viewsDir, view) + + shardMap, err := roaringMapOfShards(roaringViewPath) + if err != nil { + return nil, errors.Wrapf(err, "reading view path directory '%v'", roaringViewPath) + } + if len(shardMap) == 0 { + //vv("roaringGetFieldView2Shards C) SAVED SPACE! field '%v' view '%v' had no shards", field, view) + continue + } + + ss := newShardSetFromMap(shardMap) + fv := txkey.FieldView{Field: field, View: view} + vs.addViewShardSet(fv, ss) + + //vv("roaringGetFieldView2Shards C) added field '%v' view '%v' with shards '%#v'", field, view, ss.shards) + } + } + return +} + +// inefficient for roaring. Instead use the roaringGetFieldView2Shards() above. +func (tx *RoaringTx) GetSortedFieldViewList(idx *Index, shard uint64) (fvs []txkey.FieldView, err error) { + + // A) open the index directory + f, err := os.Open(idx.path) + if err != nil { + return nil, errors.Wrap(err, "opening directory") + } + defer f.Close() + + fieldFIs, err := f.Readdir(0) + if err != nil { + return nil, errors.Wrap(err, "reading directory") + } + + //vv("A) shard %v, opened index path '%v'", shard, idx.path) + + // B) read the name of each field under the index + for _, loopFieldFi := range fieldFIs { + fieldFI := loopFieldFi + if !fieldFI.IsDir() { + continue + } + field := fieldFI.Name() + + //vv("B) on field '%v'", field) + + fieldPath := filepath.Join(idx.path, field) + + // Skip embedded db files too. + if idx.holder.txf.IsTxDatabasePath(field) { + continue + } + viewsDir := filepath.Join(fieldPath, "views") + file, err := os.Open(viewsDir) + if os.IsNotExist(err) { + //return nil + continue + } else if err != nil { + return nil, errors.Wrapf(err, "opening view directory '%v'", viewsDir) + } + defer file.Close() + + // C) read the name of each view under the field + + viewFIs, err := file.Readdir(0) + if err != nil { + return nil, errors.Wrapf(err, "reading views directory '%v'", viewsDir) + } + for _, viewFI := range viewFIs { + + if !viewFI.IsDir() { + continue + } + view := viewFI.Name() + roaringViewPath := filepath.Join(viewsDir, view) + + shardMap, err := roaringMapOfShards(roaringViewPath) + if err != nil { + return nil, errors.Wrapf(err, "reading view path directory '%v'", roaringViewPath) + } + if len(shardMap) == 0 { + continue + } + + // once we know we have data for this shard! + if shardMap[shard] { + fv := txkey.FieldView{Field: field, View: view} + //vv("C) adding fv '%#v'", fv) + fvs = append(fvs, fv) + } + } + } + // directory stuff isn't returned in sorted order, we must sort. + sort.Slice(fvs, func(i, j int) bool { + if fvs[i].Field < fvs[j].Field { + return true + } + if fvs[i].Field > fvs[j].Field { + return false + } + return fvs[i].View < fvs[j].View + }) + return +} + //////// registrar and wrapper machinery // roaringRegistrar mirrors the machinery expected diff --git a/short_txkey/txkey.go b/short_txkey/txkey.go index 2c5af85ef..7b4cd82af 100644 --- a/short_txkey/txkey.go +++ b/short_txkey/txkey.go @@ -25,6 +25,22 @@ import ( "fmt" ) +// FieldView is here to avoid circular import. +type FieldView struct { + Field string + View string +} + +func FieldViewFromPrefix(prefix []byte) FieldView { + field, view := SplitPrefix(prefix) + return FieldView{Field: field, View: view} +} + +func FieldViewFromFullKey(fullKey []byte) FieldView { + field, view, _ := Split(fullKey) + return FieldView{Field: field, View: view} +} + // Key produces the bytes that we use as a key to query the storage/tx engine. // The roaringContainerKey argument to Key() is a container key into a roaring Container. // The return value from Key() is constructed as follows: diff --git a/stattx.go b/stattx.go index be5ea56f1..02b60d6e1 100644 --- a/stattx.go +++ b/stattx.go @@ -25,6 +25,8 @@ import ( "time" "github.com/pilosa/pilosa/v2/roaring" + txkey "github.com/pilosa/pilosa/v2/short_txkey" + //txkey "github.com/pilosa/pilosa/v2/txkey" ) // statTx is useful to profile on a @@ -675,3 +677,7 @@ func (c *statTx) Type() string { func (c *statTx) Sn() int64 { return c.b.Sn() } + +func (c *statTx) GetSortedFieldViewList(idx *Index, shard uint64) (fvs []txkey.FieldView, err error) { + return c.b.GetSortedFieldViewList(idx, shard) +} diff --git a/tx.go b/tx.go index 2d3f77fc9..771e94ff8 100644 --- a/tx.go +++ b/tx.go @@ -18,6 +18,8 @@ import ( "io" "github.com/pilosa/pilosa/v2/roaring" + txkey "github.com/pilosa/pilosa/v2/short_txkey" + //txkey "github.com/pilosa/pilosa/v2/txkey" ) // batch operations want Tx.Add(batched=doBatch), while bit-at-a-time want Tx.Add(batched=!doBatched) @@ -142,9 +144,9 @@ type Tx interface { // If the batched flag is true, then the roaring.Bitmap.AddN() is used, which does oplog batches. // If the batched flag is false, then the roaring.Bitmap.Add() is used, which does simple opTypeAdd single adds. // - // Beware: if batched is true, then changeCount will only ever be 0 or 1, + // Beware: if batched is false, then changeCount will only ever be 0 or 1, // because it calls roaring.Add(). - // If batched is false, we call roaring.DirectAddN() and then changeCount + // If batched is true, we call roaring.DirectAddN() and then changeCount // will be accurate if the changeCount is greater than 0. // // Hence: only ever call Add(batched=false) if changeCount is expected to be 0 or 1. @@ -214,6 +216,9 @@ type Tx interface { // Sn retreives the serial number of the Tx. Sn() int64 + + // GetSortedFieldViewList gets the set of FieldView(s) + GetSortedFieldViewList(idx *Index, shard uint64) (fvs []txkey.FieldView, err error) } // Closer is used by Finders diff --git a/txfactory.go b/txfactory.go index 4c2ffaa38..d8aaf3aee 100644 --- a/txfactory.go +++ b/txfactory.go @@ -30,7 +30,8 @@ import ( "github.com/pilosa/pilosa/v2/hash" "github.com/pilosa/pilosa/v2/rbf" "github.com/pilosa/pilosa/v2/roaring" - "github.com/pilosa/pilosa/v2/txkey" + txkey "github.com/pilosa/pilosa/v2/short_txkey" + //txkey "github.com/pilosa/pilosa/v2/txkey" "github.com/pkg/errors" "github.com/zeebo/blake3" ) @@ -1461,3 +1462,7 @@ func (f *TxFactory) GetDBShardPath(index string, shard uint64, idx *Index, ty tx shardPath = dbs.pathForType(ty) return } + +func (txf *TxFactory) GetFieldView2ShardsMapForIndex(idx *Index) (vs *FieldView2Shards, err error) { + return txf.dbPerShard.GetFieldView2ShardsMapForIndex(idx) +} diff --git a/txkey/txkey.go b/txkey/txkey.go index b570a8fb5..3df52a8eb 100644 --- a/txkey/txkey.go +++ b/txkey/txkey.go @@ -22,6 +22,22 @@ import ( "fmt" ) +// FieldView is here to avoid circular import. +type FieldView struct { + Field string + View string +} + +func FieldViewFromPrefix(prefix []byte) FieldView { + _, field, view, _ := SplitPrefix(prefix) + return FieldView{Field: field, View: view} +} + +func FieldViewFromFullKey(fullKey []byte) FieldView { + _, field, view, _, _ := Split(fullKey) + return FieldView{Field: field, View: view} +} + // Key produces the bytes that we use as a key to query the storage/tx engine. // The roaringContainerKey argument to Key() is a container key into a roaring Container. // The return value from Key() is constructed as follows: diff --git a/view.go b/view.go index 0929827d6..15173f5e4 100644 --- a/view.go +++ b/view.go @@ -127,8 +127,76 @@ func (v *view) removeKnownShard(shard uint64) { _, _ = v.knownShards.Remove(shard) } -// open opens and initializes the view. -func (v *view) open() error { +// openWithShardSet opens the view. Importantly, it +// only opens the fragments that have data. This saves +// a ton of time. If you have no data and want a new +// view, call view.openEmpty(). +func (v *view) openWithShardSet(ss *shardSet) error { + if v.knownShards == nil { + v.knownShards = roaring.NewSliceBitmap() + } + + // Never keep a cache for field views. + if strings.HasPrefix(v.name, viewBSIGroupPrefix) { + v.cacheType = CacheTypeNone + } + + shards := ss.CloneMaybe() + + var frags []*fragment + for shard := range shards { + frag := v.newFragment(shard) + frags = append(frags, frag) + frag.RowAttrStore = v.rowAttrStore + v.fragments[frag.shard] = frag + } + + nGoro := runtime.NumCPU() + if v.idx.holder.txf.TxType() != "roaring" { + nGoro = nGoro / 4 + } + if nGoro < 4 { + nGoro = 4 + } + pj := newParallelJobs(nGoro) + for i := range frags { + // create a new variable frag on each time through + // the loop (instead of i, frag := range frags) + // so that the closure run on the + // goroutine has its own variable. + frag := frags[i] + accepted := pj.run(func(worker int) error { + if err := frag.Open(); err != nil { + return fmt.Errorf("open fragment: shard=%d, err=%s", frag.shard, err) + } + return nil + }) + if !accepted { + // have error/shutting down the pj, so stop + break + } + } + + err := pj.waitForFinish() + if err != nil { + return err + } + + // serial, not parallel, because no locking inside addKnownShard at the moment. + // TODO(jea): is this slow on a cluster? can we optimize it + // by running it on a goroutine in the background? + for shard := range shards { + v.addKnownShard(shard) + } + + _ = testhook.Opened(v.holder.Auditor, v, nil) + v.holder.Logger.Debugf("successfully opened index/field/view: %s/%s/%s", v.index, v.field, v.name) + return nil +} + +// openEmpty opens and initializes a new view that has no +// data. If you have data already, then use view.openWithShardSet() +func (v *view) openEmpty() error { if v.knownShards == nil { v.knownShards = roaring.NewSliceBitmap() } @@ -152,10 +220,6 @@ func (v *view) open() error { v.holder.Logger.Debugf("open fragments for index/field/view: %s/%s/%s", v.index, v.field, v.name) - if err := v.openFragmentsInTx(); err != nil { - return errors.Wrap(err, "opening fragments") - } - return nil }(); err != nil { v.close() @@ -169,63 +233,6 @@ func (v *view) open() error { var workQueue = make(chan struct{}, runtime.NumCPU()*2) -// replaces v.openFragments() with Tx generic code. -func (v *view) openFragmentsInTx() error { - - shards, err := v.holder.txf.GetShardsForIndex(v.idx, v.path, false) - if err != nil { - return errors.Wrap(err, "DBPerShardGetShardsForIndex()") - } - eg, ctx := errgroup.WithContext(context.Background()) - var mu sync.Mutex - - shardCh := make(chan uint64, len(shards)) - for shard := range shards { - shardCh <- shard - } - -shardLoop: - for range shards { - select { - case <-ctx.Done(): - break shardLoop - default: - - workQueue <- struct{}{} - eg.Go(func() error { - defer func() { - <-workQueue - }() - - var shard uint64 - select { - case shard = <-shardCh: - default: - return nil // no more work - } - // these are frequent and so can expensive. Comment in only if you are actually debugging stuff. - //v.holder.Logger.Debugf("open index/field/view/fragment: %s/%s/%s/%d", v.index, v.field, v.name, shard) - - frag := v.newFragment(shard) - if err := frag.Open(); err != nil { - return fmt.Errorf("open fragment: shard=%d, err=%s", frag.shard, err) - } - frag.RowAttrStore = v.rowAttrStore - - // as above, this can be expensive, use sparely. - //v.holder.Logger.Debugf("add index/field/view/fragment to view.fragments: %s/%s/%s/%d", v.index, v.field, v.name, shard) - - mu.Lock() - v.fragments[frag.shard] = frag - v.addKnownShard(shard) - mu.Unlock() - return nil - }) - } - } - return eg.Wait() -} - // close closes the view and its fragments. func (v *view) close() error { v.mu.Lock() diff --git a/view_internal_test.go b/view_internal_test.go index 9076fb788..2b62ed04b 100644 --- a/view_internal_test.go +++ b/view_internal_test.go @@ -45,7 +45,7 @@ func mustOpenView(tb testing.TB, index, field, name string) *view { v := newView(h, path, index, field, name, fo) v.idx = idx - if err := v.open(); err != nil { + if err := v.openEmpty(); err != nil { panic(err) } v.rowAttrStore = &memAttrStore{