mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-07 00:55:55 +00:00
Merge pull request #850 from molecula/db_has_data_rb
blue_green verification and migration capabilities.
This commit is contained in:
jaten-molecula
GitHub
commit
75b01ba87d
9 changed files with 290 additions and 328 deletions
145
dbshard.go
145
dbshard.go
|
|
@ -22,7 +22,6 @@ import (
|
|||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/pkg/errors"
|
||||
)
|
||||
|
|
@ -192,6 +191,11 @@ func (dbs *DBShard) DeleteDBPath() (err error) {
|
|||
return
|
||||
}
|
||||
|
||||
type flatkey struct {
|
||||
index string
|
||||
shard uint64
|
||||
}
|
||||
|
||||
type DBPerShard struct {
|
||||
Mu sync.Mutex
|
||||
|
||||
|
|
@ -201,12 +205,13 @@ type DBPerShard struct {
|
|||
|
||||
// just flat, not buried within the Node heirarchy.
|
||||
// Easily see how many we have.
|
||||
Flatmap map[*DBShard]struct{}
|
||||
Flatmap map[flatkey]*DBShard
|
||||
|
||||
types []txtype
|
||||
hasRoaring bool
|
||||
|
||||
txf *TxFactory
|
||||
txf *TxFactory
|
||||
holder *Holder
|
||||
|
||||
// which of our types is not-roaring, since
|
||||
// roaring doesn't keep a list of open Tx sn.
|
||||
|
|
@ -215,13 +220,13 @@ type DBPerShard struct {
|
|||
}
|
||||
|
||||
// HasData returns true if the database has at least one key.
|
||||
// For roaring it returns the number of fragments stored.
|
||||
// For roaring it returns true if we a fragment stored.
|
||||
// The `which` argument is the index into the per.W slice. 0 for blue, 1 for green.
|
||||
// If you pass 1, be sure you have a blue-green configuration.
|
||||
func (per *DBPerShard) HasData(which int) (hasData bool, err error) {
|
||||
// has to aggregate across all available DBShard for each index and shard.
|
||||
|
||||
for v := range per.Flatmap {
|
||||
for _, v := range per.Flatmap {
|
||||
hasData, err = v.W[which].HasData()
|
||||
if err != nil {
|
||||
return
|
||||
|
|
@ -234,13 +239,32 @@ func (per *DBPerShard) HasData(which int) (hasData bool, err error) {
|
|||
}
|
||||
|
||||
func (per *DBPerShard) ListOpenString() (r string) {
|
||||
for v := range per.Flatmap {
|
||||
for _, v := range per.Flatmap {
|
||||
r += v.HolderPath + " -> " + v.W[per.useOpenList].OpenListString() + "\n"
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (txf *TxFactory) NewDBPerShard(types []txtype, holderDir string) (d *DBPerShard) {
|
||||
func (per *DBPerShard) LoadExistingDBs() (err error) {
|
||||
idxs := per.holder.Indexes()
|
||||
|
||||
for _, idx := range idxs {
|
||||
|
||||
sos, err := per.txf.GetShardsForIndex(idx, "", true)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, shard := range sos {
|
||||
_, err := per.GetDBShard(idx.name, shard, idx)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "DBPerShard.LoadExistingDBs GetDBShard()")
|
||||
}
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (txf *TxFactory) NewDBPerShard(types []txtype, holderDir string, holder *Holder) (d *DBPerShard) {
|
||||
|
||||
useOpenList := 0
|
||||
hasRoaring := false
|
||||
|
|
@ -261,8 +285,9 @@ func (txf *TxFactory) NewDBPerShard(types []txtype, holderDir string) (d *DBPerS
|
|||
d = &DBPerShard{
|
||||
types: types,
|
||||
HolderDir: holderDir,
|
||||
holder: holder,
|
||||
dbh: NewDBHolder(),
|
||||
Flatmap: make(map[*DBShard]struct{}),
|
||||
Flatmap: make(map[flatkey]*DBShard),
|
||||
txf: txf,
|
||||
useOpenList: useOpenList,
|
||||
hasRoaring: hasRoaring,
|
||||
|
|
@ -327,7 +352,9 @@ func (per *DBPerShard) DeleteFragment(index, field, view string, shard uint64, f
|
|||
|
||||
idx := per.txf.holder.Index(index)
|
||||
dbs, err := per.GetDBShard(index, shard, idx)
|
||||
panicOn(err)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return dbs.DeleteFragment(index, field, view, shard, frag)
|
||||
}
|
||||
|
||||
|
|
@ -357,6 +384,7 @@ func (dbs *DBShard) DumpAll() {
|
|||
func (per *DBPerShard) DumpAll() {
|
||||
per.Mu.Lock()
|
||||
defer per.Mu.Unlock()
|
||||
|
||||
found1 := false
|
||||
for _, dbi := range per.dbh.Index {
|
||||
for _, dbs := range dbi.Shard {
|
||||
|
|
@ -447,7 +475,7 @@ func (per *DBPerShard) GetDBShard(index string, shard uint64, idx *Index) (dbs *
|
|||
w.SetHolder(h)
|
||||
dbs.Open = true
|
||||
if w != nil && len(dbs.W) == 0 {
|
||||
per.Flatmap[dbs] = struct{}{}
|
||||
per.Flatmap[flatkey{index: index, shard: shard}] = dbs
|
||||
}
|
||||
dbs.W = append(dbs.W, w)
|
||||
}
|
||||
|
|
@ -464,7 +492,7 @@ func (per *DBPerShard) Del(dbs *DBShard) (err error) {
|
|||
return
|
||||
}
|
||||
panicOn(dbs.DeleteDBPath())
|
||||
delete(per.Flatmap, dbs)
|
||||
delete(per.Flatmap, flatkey{index: dbs.Index, shard: dbs.Shard})
|
||||
|
||||
// delete from the heirarchy
|
||||
delete(dbs.ParentDBIndex.Shard, dbs.Shard)
|
||||
|
|
@ -485,12 +513,14 @@ func (per *DBPerShard) Close() (err error) {
|
|||
}
|
||||
|
||||
// DBPerShardGetShardsForIndex returns the shards for idx.
|
||||
func (f *TxFactory) GetShardsForIndex(idx *Index, roaringViewPath string) (sliceOfShards []uint64, err error) {
|
||||
// If requireData, we open the database and see that it has a key, rather
|
||||
// than assume that the database file presence is enough.
|
||||
func (f *TxFactory) GetShardsForIndex(idx *Index, roaringViewPath string, requireData bool) (sliceOfShards []uint64, err error) {
|
||||
|
||||
var shards [][]uint64
|
||||
for _, ty := range f.types {
|
||||
var slc []uint64
|
||||
slc, err = f.dbPerShard.TypedDBPerShardGetShardsForIndex(ty, idx, roaringViewPath)
|
||||
slc, err = f.dbPerShard.TypedDBPerShardGetShardsForIndex(ty, idx, roaringViewPath, requireData)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
|
@ -500,32 +530,20 @@ func (f *TxFactory) GetShardsForIndex(idx *Index, roaringViewPath string) (slice
|
|||
if n != 1 && n != 2 {
|
||||
panic(fmt.Sprintf("internal error. only green or blue/green supported. we see types len %v", n))
|
||||
}
|
||||
if !f.blueGreenOff && n == 2 {
|
||||
// this is a blue green check which cannot live inside Tx because we don't know the shard yet.
|
||||
// Therefore it has to be above Tx, since we are getting all the shards to choose from here.
|
||||
//
|
||||
// But, we still want to check for blue-green consistency. In fact, this was written
|
||||
// in response to an issue with balancing/re-balancing shards being different
|
||||
// over the cluster of nodes between blue and green.
|
||||
|
||||
b := sliceToMap(shards[0])
|
||||
g := sliceToMap(shards[1])
|
||||
blueMinusGreenDiff := mapDiff(b, g)
|
||||
greenMinusBlueDiff := mapDiff(g, b)
|
||||
if len(blueMinusGreenDiff) == 0 && len(greenMinusBlueDiff) == 0 {
|
||||
// ok
|
||||
} else {
|
||||
vv("blue[%v] and green[%v] have different shards for index '%v': blueMinusGreenDiff: %v, greenMinusBlueDiff: %v; blueShards='%v', greenShards='%v'; idx.path='%v'", f.types[0].String(), f.types[1].String(), idx.name, blueMinusGreenDiff, greenMinusBlueDiff, asInts(shards[0]), asInts(shards[1]), idx.path)
|
||||
panic(fmt.Sprintf("blue[%v] and green[%v] have different shards for index '%v': blueMinusGreenDiff: %v, greenMinusBlueDiff: %v; blueShards='%v', greenShards='%v'; idx.path='%v'", f.types[0].String(), f.types[1].String(), idx.name, blueMinusGreenDiff, greenMinusBlueDiff, asInts(shards[0]), asInts(shards[1]), idx.path))
|
||||
}
|
||||
}
|
||||
// Note: we don't actually know when the blue call and when the green call comes
|
||||
// through here. So if we are deleting a shard, we will see a difference earlier
|
||||
// in one than the other. TestAPI_ClearFlagForImportAndImportValues for example.
|
||||
// Hence we cannot do a blue-green check here for matching shards.
|
||||
|
||||
// If we are populating blue from green, it does matter that we return green.
|
||||
return shards[n-1], nil
|
||||
}
|
||||
|
||||
// if roaringViewPath is "" then for ty == roaringTxn we go to disk to discover
|
||||
// all the view paths under idx for type ty.
|
||||
func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, roaringViewPath string) (sliceOfShards []uint64, err error) {
|
||||
// requireData means open the database file and verify that at least one key is set.
|
||||
func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, roaringViewPath string, requireData bool) (sliceOfShards []uint64, err error) {
|
||||
|
||||
if ty == roaringTxn {
|
||||
rx := &RoaringTx{
|
||||
|
|
@ -549,7 +567,6 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r
|
|||
return rx.SliceOfShards("", "", "", roaringViewPath)
|
||||
}
|
||||
requiredSuffix := ty.FileSuffix()
|
||||
//path := idx.Path()
|
||||
path := per.prefixForType(idx, ty)
|
||||
|
||||
ignoreEmpty := false
|
||||
|
|
@ -558,7 +575,9 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r
|
|||
panicOn(err)
|
||||
|
||||
for _, nm := range dbf {
|
||||
|
||||
base := filepath.Base(nm)
|
||||
|
||||
splt := strings.Split(base, requiredSuffix)
|
||||
if len(splt) != 2 {
|
||||
panic(fmt.Sprintf("should have 2 parts: nm='%v', base(nm)='%v'; requiredSuffix='%v'", nm, base, requiredSuffix))
|
||||
|
|
@ -569,21 +588,57 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r
|
|||
if !strings.HasPrefix(prefix, shardPrefix) {
|
||||
continue
|
||||
}
|
||||
|
||||
// Parse filename into integer.
|
||||
shard, err := strconv.ParseUint(prefix[lenOfShardPrefix:], 10, 64)
|
||||
if err != nil {
|
||||
panicOn(err)
|
||||
continue
|
||||
}
|
||||
sliceOfShards = append(sliceOfShards, shard)
|
||||
|
||||
// exclude those without data?
|
||||
hasData := false
|
||||
|
||||
if requireData {
|
||||
hasData, err = per.TypedIndexShardHasData(ty, idx, shard)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if hasData {
|
||||
sliceOfShards = append(sliceOfShards, shard)
|
||||
}
|
||||
} else {
|
||||
// file presence is enough
|
||||
sliceOfShards = append(sliceOfShards, shard)
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (per *DBPerShard) TypedIndexShardHasData(ty txtype, idx *Index, shard uint64) (hasData bool, err error) {
|
||||
whichty := 0
|
||||
if len(per.types) == 2 {
|
||||
if ty == per.types[1] {
|
||||
whichty = 1
|
||||
}
|
||||
}
|
||||
if ty != per.types[whichty] {
|
||||
return
|
||||
}
|
||||
|
||||
// make the dbs if it doesn't get exist
|
||||
dbs, err := per.GetDBShard(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()))
|
||||
}
|
||||
|
||||
return dbs.W[whichty].HasData()
|
||||
}
|
||||
|
||||
func listDirUnderDir(root string, includeRoot bool, requiredSuffix string, ignoreEmpty bool) (files []string, err error) {
|
||||
//vv("listDirUnderDir(root ='%v', suffix='%v')", root, requiredSuffix)
|
||||
|
||||
if !dirExists(root) {
|
||||
//vv("warning: listFilesUnderDir error: root directory '%v' not found", root)
|
||||
//return nil, fmt.Errorf("listFilesUnderDir error: root directory '%v' not found", root)
|
||||
return
|
||||
}
|
||||
n := len(root) + 1
|
||||
|
|
@ -649,9 +704,10 @@ func listDirUnderDir(root string, includeRoot bool, requiredSuffix string, ignor
|
|||
// populateBlueFromGreen().
|
||||
//
|
||||
func (dbs *DBShard) populateBlueFromGreen() (err error) {
|
||||
|
||||
n := len(dbs.W)
|
||||
if n != 2 {
|
||||
panic(fmt.Sprintf("copyGreenToBlue did not find 2 open DBs: have %v", n))
|
||||
panic(fmt.Sprintf("populateBlueFromGreen did not find 2 open DBs: have %v", n))
|
||||
}
|
||||
|
||||
dest := dbs.W[0] // blue
|
||||
|
|
@ -674,7 +730,12 @@ func (dbs *DBShard) populateBlueFromGreen() (err error) {
|
|||
view := vw.name
|
||||
citer, _, err := readtx.ContainerIterator(dbs.Index, field, view, dbs.Shard, 0)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "DBShard.copyGreenToBlue readtx.ContainerIterator")
|
||||
// might be an empty fragment. If so, let's not freak out.
|
||||
if strings.HasPrefix(err.Error(), "fragment not found") {
|
||||
continue
|
||||
} else {
|
||||
return errors.Wrap(err, "DBShard.populateBlueFromGreen readtx.ContainerIterator")
|
||||
}
|
||||
}
|
||||
|
||||
for citer.Next() {
|
||||
|
|
@ -682,7 +743,7 @@ func (dbs *DBShard) populateBlueFromGreen() (err error) {
|
|||
err := writetx.PutContainer(dbs.Index, field, view, dbs.Shard, ckey, rc)
|
||||
if err != nil {
|
||||
citer.Close()
|
||||
return errors.Wrap(err, "DBShard.copyGreenToBlue writetx.PutContainer")
|
||||
return errors.Wrap(err, "DBShard.populateBlueFromGreen writetx.PutContainer")
|
||||
}
|
||||
}
|
||||
citer.Close()
|
||||
|
|
@ -696,7 +757,7 @@ func (dbs *DBShard) populateBlueFromGreen() (err error) {
|
|||
}
|
||||
|
||||
// verifyBlueEqualsGreen checks that blue and green are identical.
|
||||
func (dbs *DBShard) verifyBlueEqualsGreen(numCtVerified *int64) (err error) {
|
||||
func (dbs *DBShard) verifyBlueEqualsGreen() (err error) {
|
||||
|
||||
n := len(dbs.W)
|
||||
if n != 2 {
|
||||
|
|
@ -773,8 +834,6 @@ func (dbs *DBShard) verifyBlueEqualsGreen(numCtVerified *int64) (err error) {
|
|||
"shard: '%v' ckey: '%v' nHotGreen= %v nHotBlue= %v ; BitwiseCompare response: '%v'",
|
||||
dbs.Index, field, view, dbs.Shard, greenCkey, nGreen, nBlue, err))
|
||||
}
|
||||
atomic.AddInt64(numCtVerified, 1)
|
||||
|
||||
}
|
||||
if bCiter.Next() {
|
||||
blueCkey, _ := bCiter.Value()
|
||||
|
|
|
|||
|
|
@ -21,6 +21,8 @@ import (
|
|||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/pilosa/pilosa/v2/rbf"
|
||||
)
|
||||
|
||||
// Shard per db evaluation
|
||||
|
|
@ -73,19 +75,22 @@ func Test_DBPerShard_GetShardsForIndex_LocalOnly(t *testing.T) {
|
|||
defer os.Setenv("PILOSA_TXSRC", orig) // must restore or will mess up other tests!
|
||||
|
||||
for _, src := range []string{"lmdb", "roaring", "badger", "rbf"} {
|
||||
makeSampleRoaringDir(tmpdir, src, 0)
|
||||
|
||||
os.Setenv("PILOSA_TXSRC", src)
|
||||
|
||||
// must make Holder AFTER setting src.
|
||||
holder := NewHolder(tmpdir, nil)
|
||||
|
||||
makeSampleRoaringDir(tmpdir, src, 1, holder)
|
||||
|
||||
idx, err := NewIndex(holder, tmpdir, "rick")
|
||||
panicOn(err)
|
||||
estd := "rick/_exists/views/standard"
|
||||
std := "rick/f/views/standard"
|
||||
|
||||
sos, err := holder.txf.GetShardsForIndex(idx, tmpdir+sep+std)
|
||||
sos, err := holder.txf.GetShardsForIndex(idx, tmpdir+sep+std, false)
|
||||
panicOn(err)
|
||||
|
||||
for _, shard := range []uint64{93, 223, 221, 215, 219, 217} {
|
||||
if !inSlice(sos, shard) {
|
||||
panic(fmt.Sprintf("missing shard=%v from sos='%#v'", shard, sos))
|
||||
|
|
@ -93,7 +98,7 @@ func Test_DBPerShard_GetShardsForIndex_LocalOnly(t *testing.T) {
|
|||
}
|
||||
if src == "roaring" {
|
||||
// check estd too
|
||||
sos, err = holder.txf.GetShardsForIndex(idx, tmpdir+sep+estd)
|
||||
sos, err = holder.txf.GetShardsForIndex(idx, tmpdir+sep+estd, false)
|
||||
panicOn(err)
|
||||
for _, shard := range []uint64{93, 223, 221, 215, 219, 217} {
|
||||
if !inSlice(sos, shard) {
|
||||
|
|
@ -137,67 +142,63 @@ rick/_exists/views/standard/fragments/219
|
|||
rick/_exists/views/standard/fragments/223
|
||||
`,
|
||||
"lmdb": `
|
||||
rick.index.txstores@@@/store-lmdb@@/shard.0219-lmdb@/data.mdb
|
||||
rick.index.txstores@@@/store-lmdb@@/shard.0219-lmdb@/lock.mdb
|
||||
rick.index.txstores@@@/store-lmdb@@/shard.0093-lmdb@/data.mdb
|
||||
rick.index.txstores@@@/store-lmdb@@/shard.0093-lmdb@/lock.mdb
|
||||
rick.index.txstores@@@/store-lmdb@@/shard.0223-lmdb@/data.mdb
|
||||
rick.index.txstores@@@/store-lmdb@@/shard.0223-lmdb@/lock.mdb
|
||||
rick.index.txstores@@@/store-lmdb@@/shard.0215-lmdb@/data.mdb
|
||||
rick.index.txstores@@@/store-lmdb@@/shard.0215-lmdb@/lock.mdb
|
||||
rick.index.txstores@@@/store-lmdb@@/shard.0217-lmdb@/data.mdb
|
||||
rick.index.txstores@@@/store-lmdb@@/shard.0217-lmdb@/lock.mdb
|
||||
rick.index.txstores@@@/store-lmdb@@/shard.0221-lmdb@/data.mdb
|
||||
rick.index.txstores@@@/store-lmdb@@/shard.0221-lmdb@/lock.mdb
|
||||
rick.index.txstores@@@/store-lmdb@@/shard.0093-lmdb@
|
||||
rick.index.txstores@@@/store-lmdb@@/shard.0215-lmdb@
|
||||
rick.index.txstores@@@/store-lmdb@@/shard.0217-lmdb@
|
||||
rick.index.txstores@@@/store-lmdb@@/shard.0219-lmdb@
|
||||
rick.index.txstores@@@/store-lmdb@@/shard.0221-lmdb@
|
||||
rick.index.txstores@@@/store-lmdb@@/shard.0223-lmdb@
|
||||
`,
|
||||
"badger": `
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0219-badgerdb@/000000.vlog
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0219-badgerdb@/KEYREGISTRY
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0219-badgerdb@/MANIFEST
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0219-badgerdb@/LOCK
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0221-badgerdb@/000000.vlog
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0221-badgerdb@/KEYREGISTRY
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0221-badgerdb@/MANIFEST
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0221-badgerdb@/LOCK
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0223-badgerdb@/000000.vlog
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0223-badgerdb@/KEYREGISTRY
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0223-badgerdb@/MANIFEST
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0223-badgerdb@/LOCK
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0093-badgerdb@/000000.vlog
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0093-badgerdb@/KEYREGISTRY
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0093-badgerdb@/MANIFEST
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0093-badgerdb@/LOCK
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0217-badgerdb@/000000.vlog
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0217-badgerdb@/KEYREGISTRY
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0217-badgerdb@/MANIFEST
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0217-badgerdb@/LOCK
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0215-badgerdb@/000000.vlog
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0215-badgerdb@/KEYREGISTRY
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0215-badgerdb@/MANIFEST
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0215-badgerdb@/LOCK
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0093-badgerdb@
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0215-badgerdb@
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0217-badgerdb@
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0219-badgerdb@
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0221-badgerdb@
|
||||
rick.index.txstores@@@/store-badgerdb@@/shard.0223-badgerdb@
|
||||
`,
|
||||
"rbf": `
|
||||
rick.index.txstores@@@/store-rbfdb@@/shard.0223-rbfdb@/wal/0000000000000001.wal
|
||||
rick.index.txstores@@@/store-rbfdb@@/shard.0223-rbfdb@/data
|
||||
rick.index.txstores@@@/store-rbfdb@@/shard.0093-rbfdb@/wal/0000000000000001.wal
|
||||
rick.index.txstores@@@/store-rbfdb@@/shard.0093-rbfdb@/data
|
||||
rick.index.txstores@@@/store-rbfdb@@/shard.0217-rbfdb@/wal/0000000000000001.wal
|
||||
rick.index.txstores@@@/store-rbfdb@@/shard.0217-rbfdb@/data
|
||||
rick.index.txstores@@@/store-rbfdb@@/shard.0215-rbfdb@/wal/0000000000000001.wal
|
||||
rick.index.txstores@@@/store-rbfdb@@/shard.0215-rbfdb@/data
|
||||
rick.index.txstores@@@/store-rbfdb@@/shard.0221-rbfdb@/wal/0000000000000001.wal
|
||||
rick.index.txstores@@@/store-rbfdb@@/shard.0221-rbfdb@/data
|
||||
rick.index.txstores@@@/store-rbfdb@@/shard.0219-rbfdb@/wal/0000000000000001.wal
|
||||
rick.index.txstores@@@/store-rbfdb@@/shard.0219-rbfdb@/data
|
||||
rick.index.txstores@@@/store-rbfdb@@/shard.0093-rbfdb@
|
||||
rick.index.txstores@@@/store-rbfdb@@/shard.0215-rbfdb@
|
||||
rick.index.txstores@@@/store-rbfdb@@/shard.0217-rbfdb@
|
||||
rick.index.txstores@@@/store-rbfdb@@/shard.0219-rbfdb@
|
||||
rick.index.txstores@@@/store-rbfdb@@/shard.0221-rbfdb@
|
||||
rick.index.txstores@@@/store-rbfdb@@/shard.0223-rbfdb@
|
||||
`,
|
||||
}
|
||||
|
||||
func makeSampleRoaringDir(root, txsrc string, minBytes int) {
|
||||
func makeSampleRoaringDir(root, txsrc string, minBytes int, h *Holder) {
|
||||
|
||||
index := "rick"
|
||||
shards := []uint64{0, 93, 215, 217, 219, 221, 223}
|
||||
fns := strings.Split(sampleRoaringDirList[txsrc], "\n")
|
||||
for _, fn := range fns {
|
||||
for i, fn := range fns {
|
||||
if fn == "" {
|
||||
continue
|
||||
}
|
||||
var shard uint64
|
||||
if txsrc != "roaring" {
|
||||
// only have shards for the non-roaring
|
||||
shard = shards[i]
|
||||
}
|
||||
switch txsrc {
|
||||
case "lmdb":
|
||||
makeLMDBtestDB(root+sep+fn, h, shard)
|
||||
// also have to make the DBShard in our in-memory tree,
|
||||
// or else the search won't find it because
|
||||
// DBPerShard won't know anything about it.
|
||||
helperCreateDBShard(h, index, shard)
|
||||
continue
|
||||
case "badger":
|
||||
makeBadgertestDB(root+sep+fn, h, shard)
|
||||
helperCreateDBShard(h, index, shard)
|
||||
continue
|
||||
case "rbf":
|
||||
makeRBFtestDB(root+sep+fn, h, shard)
|
||||
helperCreateDBShard(h, index, shard)
|
||||
continue
|
||||
}
|
||||
|
||||
path := root + sep + filepath.Dir(fn)
|
||||
panicOn(os.MkdirAll(path, 0755))
|
||||
fd, err := os.Create(root + sep + fn)
|
||||
|
|
@ -209,3 +210,47 @@ func makeSampleRoaringDir(root, txsrc string, minBytes int) {
|
|||
fd.Close()
|
||||
}
|
||||
}
|
||||
|
||||
func helperCreateDBShard(h *Holder, index string, shard uint64) {
|
||||
idx, err := h.CreateIndexIfNotExists(index, IndexOptions{})
|
||||
panicOn(err)
|
||||
dbs, err := h.txf.dbPerShard.GetDBShard(index, shard, idx)
|
||||
panicOn(err)
|
||||
_ = dbs
|
||||
}
|
||||
|
||||
func makeLMDBtestDB(path string, h *Holder, shard uint64) {
|
||||
i := uint64(1)
|
||||
w, _ := mustOpenEmptyLMDBWrapper(path)
|
||||
LMDBMustSetBitvalue(w, "index", "field", "view", shard, i)
|
||||
w.Close()
|
||||
|
||||
}
|
||||
|
||||
func makeBadgertestDB(path string, h *Holder, shard uint64) {
|
||||
i := uint64(1)
|
||||
w, _ := mustOpenEmptyBadgerWrapper(path)
|
||||
badgerDBMustSetBitvalue(w, "index", "field", "view", shard, i)
|
||||
w.Close()
|
||||
}
|
||||
|
||||
func makeRBFtestDB(path string, h *Holder, shard uint64) {
|
||||
i := uint64(1)
|
||||
|
||||
db := rbf.NewDB(path)
|
||||
err := db.Open()
|
||||
panicOn(err)
|
||||
defer db.Close()
|
||||
|
||||
tx, err := db.Begin(true)
|
||||
panicOn(err)
|
||||
|
||||
err = tx.CreateBitmap("x")
|
||||
panicOn(err)
|
||||
|
||||
_, err = tx.Add("x", i)
|
||||
panicOn(err)
|
||||
|
||||
err = tx.Commit()
|
||||
panicOn(err)
|
||||
}
|
||||
|
|
|
|||
1
go.mod
1
go.mod
|
|
@ -12,7 +12,6 @@ require (
|
|||
github.com/codahale/hdrhistogram v0.0.0-20161010025455-3a0bb77429bd // indirect
|
||||
github.com/davecgh/go-spew v1.1.1
|
||||
github.com/dgraph-io/badger/v2 v2.0.1-rc1.0.20200709123515-8e896a7af361
|
||||
github.com/glycerine/idem v0.0.0-20190127113923-7a8083893311
|
||||
github.com/glycerine/lmdb-go v1.9.32
|
||||
github.com/go-ole/go-ole v1.2.4 // indirect
|
||||
github.com/gogo/protobuf v1.2.1
|
||||
|
|
|
|||
135
holdbkg.go
135
holdbkg.go
|
|
@ -1,135 +0,0 @@
|
|||
// Copyright 2017 Pilosa Corp.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package pilosa
|
||||
|
||||
import (
|
||||
"github.com/glycerine/idem"
|
||||
)
|
||||
|
||||
// holderBackgroundGoro avoids the deadlock versus race dilmena
|
||||
// when creating a new index. The troubles were in having
|
||||
// the field/view code try to inform the Holder of the indexes
|
||||
// from a platoon of worker goroutines and tests; see index.go:273 inside
|
||||
// Index.openFields(). There calling i.holder.addIndex(i) instead
|
||||
// of locking cleans things up alot. There's no need anymore
|
||||
// to track which goroutines are holding the Holder's
|
||||
// mu lock.
|
||||
type holderBackgroundGoro struct {
|
||||
h *Holder
|
||||
halt *idem.Halter
|
||||
reqIndexCh chan *indexReq
|
||||
setIdxCh chan *Index
|
||||
getAllCh chan *indexReq
|
||||
delIdxCh chan string
|
||||
}
|
||||
|
||||
// indexReq is used to ask the holderBackgrounGoro
|
||||
// for index information.
|
||||
type indexReq struct {
|
||||
// basic request to map an index name to *Index
|
||||
index string
|
||||
idx *Index
|
||||
|
||||
// double duty as a getAll indexes request
|
||||
getAll bool
|
||||
all []*Index
|
||||
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
func newIndexReq(index string) *indexReq {
|
||||
return &indexReq{
|
||||
index: index,
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
func (b *holderBackgroundGoro) index(index string) *Index {
|
||||
if b == nil {
|
||||
// utils_internal_test.go:448 from
|
||||
// TestCluster_ResizeStates/Multiple_nodes,_with_data
|
||||
// wants to pass a nil b/nil Holder. sigh. don't panic.
|
||||
return nil
|
||||
}
|
||||
req := newIndexReq(index)
|
||||
b.reqIndexCh <- req
|
||||
<-req.done
|
||||
return req.idx
|
||||
}
|
||||
|
||||
func (b *holderBackgroundGoro) isClosed() bool {
|
||||
select {
|
||||
case <-b.halt.Done.Chan:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func (b *holderBackgroundGoro) start() {
|
||||
go func() {
|
||||
defer b.halt.Done.Close()
|
||||
for {
|
||||
select {
|
||||
case <-b.halt.ReqStop.Chan:
|
||||
return
|
||||
|
||||
case r := <-b.reqIndexCh:
|
||||
r.idx = b.h.indexesOwnedByBkgr[r.index]
|
||||
close(r.done)
|
||||
|
||||
case idx := <-b.setIdxCh:
|
||||
b.h.indexesOwnedByBkgr[idx.Name()] = idx
|
||||
// atomic operation, client doesn't need to wait on done,
|
||||
// so don't take the time to do another channel, just leave done open.
|
||||
|
||||
case target := <-b.delIdxCh:
|
||||
delete(b.h.indexesOwnedByBkgr, target)
|
||||
// atomic operation, client doesn't need to wait on done,
|
||||
// so don't take the time to do another channel, just leave done open.
|
||||
|
||||
case r := <-b.getAllCh:
|
||||
r.all = make([]*Index, 0, len(b.h.indexesOwnedByBkgr))
|
||||
for _, index := range b.h.indexesOwnedByBkgr {
|
||||
r.all = append(r.all, index)
|
||||
}
|
||||
close(r.done)
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func (h *Holder) stopBkgr() {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
if h.bkgr != nil {
|
||||
h.bkgr.halt.ReqStop.Close()
|
||||
<-h.bkgr.halt.Done.Chan
|
||||
}
|
||||
}
|
||||
|
||||
func (h *Holder) newHolderBackgroundGoro() (b *holderBackgroundGoro) {
|
||||
|
||||
b = &holderBackgroundGoro{
|
||||
h: h,
|
||||
halt: idem.NewHalter(),
|
||||
reqIndexCh: make(chan *indexReq),
|
||||
setIdxCh: make(chan *Index),
|
||||
getAllCh: make(chan *indexReq),
|
||||
delIdxCh: make(chan string),
|
||||
}
|
||||
b.start()
|
||||
return b
|
||||
}
|
||||
99
holder.go
99
holder.go
|
|
@ -117,31 +117,10 @@ type Holder struct {
|
|||
|
||||
txf *TxFactory
|
||||
|
||||
bkgr *holderBackgroundGoro
|
||||
|
||||
// indexesOwnedByBkgr is owned by h.bkgr. Do not touch.
|
||||
// Only the owning holdbkg.go goroutine should query or
|
||||
// modify the indexesOwnedByBkgr map. indexesOwnedByBkgr replaces
|
||||
// the old indexes map which was the source of deadlock vs
|
||||
// race issues with a dedicated goroutine
|
||||
// associated with the Holder.
|
||||
//
|
||||
// Normally this map would live inside the holderBackgroundGoro
|
||||
// struct, but tests expect the map to
|
||||
// persist cross Holder ReOpens(), so it still lives here --
|
||||
// since a ReOpen will kill and restart the goroutine and
|
||||
// lose its state.
|
||||
//
|
||||
// Again, do not touch this directly. Use these methods instead:
|
||||
//
|
||||
// h.Index() // index name -> *Index
|
||||
// h.Indexes() // copy of the full list of *Indexes
|
||||
// h.addIndex() // add one *Index
|
||||
// h.deleteIndex() // delete one *Index
|
||||
//
|
||||
indexesOwnedByBkgr map[string]*Index
|
||||
|
||||
numCtBlueGreenVerified int64
|
||||
// a separate lock out for indexes, to avoid the deadlock/race dilema
|
||||
// on holding mu.
|
||||
imu sync.RWMutex
|
||||
indexes map[string]*Index
|
||||
}
|
||||
|
||||
// HolderOpts holds information about the holder which other things might want
|
||||
|
|
@ -235,8 +214,6 @@ func DefaultHolderConfig() *HolderConfig {
|
|||
}
|
||||
|
||||
// NewHolder returns a new instance of Holder for the given path.
|
||||
// It starts the bkgr background goroutine that provides
|
||||
// exclusive access to the indexesOwnedByBkgr map.
|
||||
func NewHolder(path string, cfg *HolderConfig) *Holder {
|
||||
if cfg == nil {
|
||||
cfg = DefaultHolderConfig()
|
||||
|
|
@ -274,13 +251,13 @@ func NewHolder(path string, cfg *HolderConfig) *Holder {
|
|||
|
||||
path: path,
|
||||
|
||||
indexesOwnedByBkgr: make(map[string]*Index),
|
||||
indexes: make(map[string]*Index),
|
||||
}
|
||||
h.bkgr = h.newHolderBackgroundGoro()
|
||||
|
||||
txf, err := NewTxFactory(cfg.Txsrc, path, h)
|
||||
panicOn(err)
|
||||
h.txf = txf
|
||||
h.txf.blueGreenOffIfRunningBlueGreen()
|
||||
|
||||
_ = testhook.Created(h.Auditor, h, nil)
|
||||
return h
|
||||
|
|
@ -571,10 +548,6 @@ func (h *Holder) Open() error {
|
|||
h.opening = true
|
||||
defer func() { h.opening = false }()
|
||||
|
||||
if h.bkgr == nil || h.bkgr.isClosed() {
|
||||
h.bkgr = h.newHolderBackgroundGoro()
|
||||
}
|
||||
|
||||
if h.txf == nil {
|
||||
txf, err := NewTxFactory(h.cfg.Txsrc, h.path, h)
|
||||
if err != nil {
|
||||
|
|
@ -673,6 +646,10 @@ func (h *Holder) Open() error {
|
|||
|
||||
_ = testhook.Opened(h.Auditor, h, nil)
|
||||
|
||||
if err := h.txf.Open(); err != nil {
|
||||
return errors.Wrap(err, "Holder.Open h.txf.Open()")
|
||||
}
|
||||
|
||||
// under blue_green, we must sync blue from green before we turn on checking.
|
||||
if err := h.txf.green2blue(h); err != nil {
|
||||
return errors.Wrap(err, "Holder.Open h.txf.green2blue(h)")
|
||||
|
|
@ -728,7 +705,6 @@ func (h *Holder) Close() error {
|
|||
if h == nil {
|
||||
return nil
|
||||
}
|
||||
defer h.stopBkgr()
|
||||
|
||||
if globalUseStatTx {
|
||||
fmt.Printf("%v\n", globalCallStats.report())
|
||||
|
|
@ -977,25 +953,24 @@ func (h *Holder) HolderPathFromIndexPath(indexPath, indexName string) string {
|
|||
|
||||
// Index returns the index by name.
|
||||
func (h *Holder) Index(name string) (idx *Index) {
|
||||
idx = h.bkgr.index(name)
|
||||
h.imu.RLock()
|
||||
idx = h.indexes[name]
|
||||
h.imu.RUnlock()
|
||||
return
|
||||
}
|
||||
|
||||
// Indexes returns a list of all indexes in the holder.
|
||||
func (h *Holder) Indexes() []*Index {
|
||||
// utils_internal_test.go:345 func (t *ClusterCluster) Close() error
|
||||
// wants to close an un-Open()-ed Holder.
|
||||
// So I guess we won't panic here.
|
||||
if h.bkgr == nil || h.bkgr.isClosed() {
|
||||
return nil
|
||||
h.imu.RLock()
|
||||
// sizing and copying has to be done under the lock to avoid
|
||||
// a logical race with a deletion/addition to indexes.
|
||||
cp := make([]*Index, 0, len(h.indexes))
|
||||
for _, idx := range h.indexes {
|
||||
cp = append(cp, idx)
|
||||
}
|
||||
req := newIndexReq("")
|
||||
req.getAll = true
|
||||
h.bkgr.getAllCh <- req
|
||||
<-req.done
|
||||
|
||||
sort.Sort(indexSlice(req.all))
|
||||
return req.all
|
||||
h.imu.RUnlock()
|
||||
sort.Sort(indexSlice(cp))
|
||||
return cp
|
||||
}
|
||||
|
||||
// CreateIndex creates an index.
|
||||
|
|
@ -1107,7 +1082,9 @@ func (h *Holder) DeleteIndex(name string) error {
|
|||
}
|
||||
|
||||
func (h *Holder) deleteIndex(index string) {
|
||||
h.bkgr.delIdxCh <- index
|
||||
h.imu.Lock()
|
||||
delete(h.indexes, index)
|
||||
h.imu.Unlock()
|
||||
}
|
||||
|
||||
// Field returns the field for an index and name.
|
||||
|
|
@ -1973,25 +1950,9 @@ func (h *Holder) Process(ctx context.Context, op HolderOperator) (err error) {
|
|||
// used by Index.openFields(), enabling Tx / Txf by telling
|
||||
// the holder about its own indexes.
|
||||
func (h *Holder) addIndex(idx *Index) {
|
||||
if idx == nil {
|
||||
panic("cannot pass nil to addIndex")
|
||||
}
|
||||
if h == nil {
|
||||
panic("cannot call addIndex on nil Holder")
|
||||
}
|
||||
if h.bkgr == nil || h.bkgr.isClosed() {
|
||||
// ugh. TestCluster_ResizeStates/Multiple_nodes,_with_data test
|
||||
// from cluster_internal_test.go:799
|
||||
// via utils_internal_test.go:450
|
||||
// gets here, with the h.mu already held.
|
||||
// so we cannot panic and complain or we mess up that test.
|
||||
// But really, we should not be calling addIndex() on
|
||||
// Holder that has not be Open()-ed.
|
||||
h.mu.Lock()
|
||||
h.bkgr = h.newHolderBackgroundGoro()
|
||||
h.mu.Unlock()
|
||||
}
|
||||
h.bkgr.setIdxCh <- idx
|
||||
h.imu.Lock()
|
||||
h.indexes[idx.name] = idx
|
||||
h.imu.Unlock()
|
||||
}
|
||||
|
||||
func (h *Holder) DumpAllShards() {
|
||||
|
|
@ -2012,10 +1973,6 @@ func (h *Holder) BeginTx(writable bool, idx *Index, shard uint64) (Tx, error) {
|
|||
return h.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard}), nil
|
||||
}
|
||||
|
||||
func (h *Holder) NumContainersBlueGreenVerified() int {
|
||||
return int(h.numCtBlueGreenVerified)
|
||||
}
|
||||
|
||||
func (h *Holder) HasRoaringData() (has bool, err error) {
|
||||
|
||||
idxs := h.Indexes()
|
||||
|
|
|
|||
|
|
@ -152,6 +152,9 @@ func (c *Cursor) CurrentPageType() int {
|
|||
|
||||
func toContainer(l leafCell, tx *Tx) (c *roaring.Container) {
|
||||
|
||||
if len(l.Data) == 0 {
|
||||
return nil
|
||||
}
|
||||
orig := l.Data
|
||||
var cpMaybe []byte
|
||||
var mapped bool
|
||||
|
|
|
|||
43
txfactory.go
43
txfactory.go
|
|
@ -493,7 +493,7 @@ func MustTxsrcToTxtype(txsrc string) (types []txtype) {
|
|||
|
||||
// NewTxFactory always opens an existing database. If you
|
||||
// want to a fresh database, os.RemoveAll on dir/name ahead of time.
|
||||
// We always store files in a subdir of dir. If we are having one
|
||||
// We always store files in a subdir of holderDir. If we are having one
|
||||
// database or many can depend on name.
|
||||
func NewTxFactory(txsrc string, holderDir string, holder *Holder) (f *TxFactory, err error) {
|
||||
types := MustTxsrcToTxtype(txsrc)
|
||||
|
|
@ -506,11 +506,17 @@ func NewTxFactory(txsrc string, holderDir string, holder *Holder) (f *TxFactory,
|
|||
if len(types) == 2 {
|
||||
f.blueGreenReg = newBlueGreenReg(types)
|
||||
}
|
||||
f.dbPerShard = f.NewDBPerShard(types, holderDir)
|
||||
f.dbPerShard = f.NewDBPerShard(types, holderDir, holder)
|
||||
|
||||
return f, err
|
||||
}
|
||||
|
||||
// Open should be called only once the index metadata is loaded
|
||||
// from Holder.Open(), so we find all of our indexes.
|
||||
func (f *TxFactory) Open() error {
|
||||
return f.dbPerShard.LoadExistingDBs()
|
||||
}
|
||||
|
||||
// Txo holds the transaction options
|
||||
type Txo struct {
|
||||
Write bool
|
||||
|
|
@ -1175,6 +1181,7 @@ func (f *TxFactory) blueHasData() (hasData bool, err error) {
|
|||
// Called by test Test_TxFactory_UpdateBlueFromGreen_OnStartup() in
|
||||
// txfactory_internal_test.go as well.
|
||||
//
|
||||
// This is a noop if we aren't running under a blue_green PILOSA_TXSRC.
|
||||
func (f *TxFactory) green2blue(holder *Holder) (err error) {
|
||||
|
||||
// Holder.Open will always call us, even without blue_green. Which is fine.
|
||||
|
|
@ -1201,32 +1208,32 @@ func (f *TxFactory) green2blue(holder *Holder) (err error) {
|
|||
for _, idx := range idxs {
|
||||
|
||||
// scan directories
|
||||
blueShards, err := f.dbPerShard.TypedDBPerShardGetShardsForIndex(blueDest, idx, "")
|
||||
blueShards, err := f.dbPerShard.TypedDBPerShardGetShardsForIndex(blueDest, idx, "", false)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, fmt.Sprintf("GetDBShard(index='%v') error fetching blueShards", idx.name))
|
||||
}
|
||||
//vv("from blueDest='%v', blueShards = '%#v'", blueDest, blueShards)
|
||||
|
||||
// scan directories
|
||||
greenShards, err := f.dbPerShard.TypedDBPerShardGetShardsForIndex(greenSrc, idx, "")
|
||||
greenShards, err := f.dbPerShard.TypedDBPerShardGetShardsForIndex(greenSrc, idx, "", true)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, fmt.Sprintf("GetDBShard(index='%v') error fetching greenShards", idx.name))
|
||||
}
|
||||
//vv("from greenSrc='%v', greenShards = '%#v'", greenSrc, greenShards)
|
||||
|
||||
diff := f.shardSliceDiff(blueShards, greenShards)
|
||||
if diff != "" {
|
||||
return fmt.Errorf("blue[%v] and green[%v] have different shards for index '%v': %v", blueDest, greenSrc, idx.name, diff)
|
||||
if verifyInsteadOfCopy {
|
||||
diff := f.shardSliceDiff(blueShards, greenShards)
|
||||
if diff != "" {
|
||||
return fmt.Errorf("verifyInsteadOfCopy true, blue[%v]=%#v and green[%v]=%#v have different shards for index '%v': '%v'; stack=\n%v", blueDest, blueShards, greenSrc, greenShards, idx.name, diff, stack())
|
||||
}
|
||||
|
||||
// can also check against meta data
|
||||
shards := idx.AvailableShards(localOnly).Slice()
|
||||
diff2 := f.shardSliceDiff(greenShards, shards)
|
||||
if diff2 != "" {
|
||||
return fmt.Errorf("green[%v] = '%#v' and meta data '%#v' have different shards for index '%v': %v", greenSrc, greenShards, shards, idx.name, diff2)
|
||||
}
|
||||
}
|
||||
|
||||
shards := idx.AvailableShards(localOnly).Slice()
|
||||
|
||||
diff2 := f.shardSliceDiff(blueShards, shards)
|
||||
if diff2 != "" {
|
||||
return fmt.Errorf("blue[%v] and meta data (from green[%v]?)have different shards for index '%v': %v", blueDest, greenSrc, idx.name, diff2)
|
||||
}
|
||||
|
||||
for _, shard := range shards {
|
||||
for _, shard := range greenShards {
|
||||
|
||||
dbs, err := f.dbPerShard.GetDBShard(idx.name, shard, idx)
|
||||
if err != nil {
|
||||
|
|
@ -1235,7 +1242,7 @@ func (f *TxFactory) green2blue(holder *Holder) (err error) {
|
|||
|
||||
if verifyInsteadOfCopy {
|
||||
// verify all containers
|
||||
err = dbs.verifyBlueEqualsGreen(&holder.numCtBlueGreenVerified)
|
||||
err = dbs.verifyBlueEqualsGreen()
|
||||
if err != nil {
|
||||
return errors.Wrap(err,
|
||||
fmt.Sprintf("dbs.verifyBlueEqualsGreen(blue='%v', "+
|
||||
|
|
|
|||
|
|
@ -112,12 +112,11 @@ func Test_TxFactory_Qcx_query_context(t *testing.T) {
|
|||
// and b) we have an easy migration mechanism, to go from one storage format to another.
|
||||
//
|
||||
func Test_TxFactory_UpdateBlueFromGreen_OnStartup(t *testing.T) {
|
||||
//t.Skip("TODO(jea) bring this back in. broken by the local vs remote shard determination for a cluster")
|
||||
|
||||
orig := os.Getenv("PILOSA_TXSRC")
|
||||
defer os.Setenv("PILOSA_TXSRC", orig) // must restore or will mess up other tests!
|
||||
|
||||
checked := []string{"lmdb", "roaring", "badger", "rbf"}
|
||||
checked := []string{"lmdb", "roaring", "rbf"}
|
||||
|
||||
for _, blue := range checked {
|
||||
for _, green := range checked {
|
||||
|
|
@ -125,6 +124,7 @@ func Test_TxFactory_UpdateBlueFromGreen_OnStartup(t *testing.T) {
|
|||
continue
|
||||
}
|
||||
blue_green := blue + "_" + green
|
||||
//vv("setting blue_green to '%v'", blue_green)
|
||||
|
||||
// =============================
|
||||
// Begin setup.
|
||||
|
|
@ -137,6 +137,7 @@ func Test_TxFactory_UpdateBlueFromGreen_OnStartup(t *testing.T) {
|
|||
t.Fatalf("creating holder: %v", err)
|
||||
}
|
||||
defer os.RemoveAll(path)
|
||||
//vv("path = %v", path)
|
||||
|
||||
// we will manually h.Close() below
|
||||
|
||||
|
|
@ -169,17 +170,26 @@ func Test_TxFactory_UpdateBlueFromGreen_OnStartup(t *testing.T) {
|
|||
testMustHaveBit(t, h, "i1", "f", 100, 200)
|
||||
testMustHaveBit(t, h, "i1", "f", 100, 12345678)
|
||||
|
||||
//vv("about to reopen; blue_green = '%v' but PILOSA_TXSRC='%v'", blue_green, os.Getenv("PILOSA_TXSRC"))
|
||||
//h.DumpAllShards()
|
||||
|
||||
//vv("after dump, about to close")
|
||||
h.Close()
|
||||
|
||||
//vv("after close, about to re-open")
|
||||
|
||||
// can we re.Open the same holder h? hopefully without a problem.
|
||||
panicOn(h.Open())
|
||||
|
||||
testMustHaveBit(t, h, "i0", "f", rowID, colID)
|
||||
//vv("h.Open() re-open worked; blue_green = '%v'; dump; with PILOSA_TXSRC='%v'", blue_green, os.Getenv("PILOSA_TXSRC"))
|
||||
//h.DumpAllShards()
|
||||
|
||||
testMustHaveBit(t, h, "i0", "f", rowID, colID) // panic here, colID 200 bit was cold.
|
||||
testMustHaveBit(t, h, "i1", "f", 100, 200)
|
||||
testMustHaveBit(t, h, "i1", "f", 100, 12345678)
|
||||
h.Close()
|
||||
|
||||
// successful re-open and then Close again of h.
|
||||
//vv("successful re-open and then Close again of h.")
|
||||
|
||||
// check that we can open a NewHolder on green, on same path, and still see our bits.
|
||||
// Because the NewHolder is the code that creates and configures TxFactory as blue_green.
|
||||
|
|
@ -221,11 +231,14 @@ func Test_TxFactory_UpdateBlueFromGreen_OnStartup(t *testing.T) {
|
|||
|
||||
//vv("about to h4.Open we should populate blue from green")
|
||||
panicOn(h4.Open())
|
||||
defer h4.Close()
|
||||
|
||||
testMustHaveBit(t, h4, "i0", "f", rowID, colID)
|
||||
testMustHaveBit(t, h4, "i1", "f", 100, 200)
|
||||
testMustHaveBit(t, h4, "i1", "f", 100, 12345678)
|
||||
|
||||
//vv("successfully verified populatingBlueFromGreen with blue_green = '%v'", blue_green)
|
||||
h4.Close()
|
||||
os.RemoveAll(path)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -234,7 +247,6 @@ func Test_TxFactory_UpdateBlueFromGreen_OnStartup(t *testing.T) {
|
|||
// go to verify it but blue has more data than green.
|
||||
// That will also cause query divergence.
|
||||
func Test_TxFactory_verifyBlueEqualsGreen(t *testing.T) {
|
||||
//t.Skip("TODO(jea) bring this back in. broken by the local vs remote shard determination for a cluster")
|
||||
|
||||
orig := os.Getenv("PILOSA_TXSRC")
|
||||
defer os.Setenv("PILOSA_TXSRC", orig) // must restore or will mess up other tests!
|
||||
|
|
@ -260,6 +272,7 @@ func Test_TxFactory_verifyBlueEqualsGreen(t *testing.T) {
|
|||
}
|
||||
defer os.RemoveAll(path)
|
||||
|
||||
//vv("on green, which is '%v'", green)
|
||||
// we will manually h.Close() below
|
||||
|
||||
// Write bits to separate indexes.
|
||||
|
|
@ -297,6 +310,8 @@ func Test_TxFactory_verifyBlueEqualsGreen(t *testing.T) {
|
|||
// open a new holder on path, just looking at blue.
|
||||
os.Setenv("PILOSA_TXSRC", blue)
|
||||
|
||||
//vv("on blue, which is '%v'", blue)
|
||||
|
||||
h3 := NewHolder(path, nil)
|
||||
panicOn(h3.Open())
|
||||
|
||||
|
|
@ -317,6 +332,8 @@ func Test_TxFactory_verifyBlueEqualsGreen(t *testing.T) {
|
|||
|
||||
os.Setenv("PILOSA_TXSRC", blue_green)
|
||||
|
||||
//vv("on blue_green, which is '%v'", blue_green)
|
||||
|
||||
// open a holder with path again, now looking at both blue and green.
|
||||
// The Holder.Open should do the migration from green, populating blue.
|
||||
h4 := NewHolder(path, nil)
|
||||
|
|
@ -330,9 +347,15 @@ func Test_TxFactory_verifyBlueEqualsGreen(t *testing.T) {
|
|||
// now open just blue, and add a bit to a new index, i2.
|
||||
os.Setenv("PILOSA_TXSRC", blue)
|
||||
|
||||
//vv("on blue, which is '%v'", blue)
|
||||
|
||||
h5 := NewHolder(path, nil)
|
||||
panicOn(h5.Open())
|
||||
testSetBit(t, h5, "i2", "f", 500, 777)
|
||||
|
||||
//vv("after adding a bit to blue, we have:")
|
||||
//h5.DumpAllShards()
|
||||
|
||||
h5.Close()
|
||||
|
||||
// now open blue_green. should get a verification failure
|
||||
|
|
@ -345,6 +368,10 @@ func Test_TxFactory_verifyBlueEqualsGreen(t *testing.T) {
|
|||
// The Holder.Open should verify blue against green and notice the extra bit.
|
||||
h6 := NewHolder(path, nil)
|
||||
err = h6.Open()
|
||||
|
||||
//vv("h6.Open() had err = '%v', PILOSA_TXSRC='%v'", err, os.Getenv("PILOSA_TXSRC"))
|
||||
//h6.DumpAllShards()
|
||||
|
||||
if err == nil {
|
||||
h6.Close()
|
||||
t.Fatalf("should have had blue-green verification fail on Holder.Open")
|
||||
|
|
|
|||
2
view.go
2
view.go
|
|
@ -173,7 +173,7 @@ 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)
|
||||
shards, err := v.holder.txf.GetShardsForIndex(v.idx, v.path, false)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "DBPerShardGetShardsForIndex()")
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue