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

pilosa: only open views with data

Browse source 
- 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.
This commit is contained in:
Jason E. Aten 2020-12-17 21:25:55 +00:00
parent 7b6c6303ce
commit 035073555a
21 changed files with 769 additions and 186 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
bluegreentx.go
View file

@ -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

21
bolt.go
View file

@ -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.

28
bolt_test.go
View file

@ -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))
}

6
catcher.go
View file

@ -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)
}

223
dbshard.go
View file

@ -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)
}

170
dbshard_internal_test.go
View file

@ -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))
}
}

94
field.go
View file

@ -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

2
fragment_internal_test.go
View file

@ -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]
}

2
holder.go
View file

@ -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")

15
index.go
View file

@ -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)
}

5
rbf.go
View file

@ -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))

1
rbf/ingest_test.go
View file

@ -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"
)

16
rbf/tx.go
View file

@ -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()
}

172
rrtx.go
View file

@ -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

16
short_txkey/txkey.go
View file

@ -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:

6
stattx.go
View file

@ -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)
}

9
tx.go
View file

@ -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

7
txfactory.go
View file

@ -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)
}

16
txkey/txkey.go
View file

@ -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:

133
view.go
View file

@ -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()

2
view_internal_test.go
View file

@ -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{