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

remove code related to pilosa-chk

Browse source
This commit is contained in:
Travis 2021-02-05 14:22:16 -06:00
parent b7d6db51a3
commit 002aee63e3
No known key found for this signature in database
GPG key ID: 37080CC2042BA34E
4 changed files with 0 additions and 1079 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

841
boltdb/translate.go
View file

@ -22,14 +22,11 @@ import (
"io/ioutil"
"os"
"path/filepath"
"sort"
"sync"
"time"
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/topology"
"github.com/pkg/errors"
"github.com/zeebo/blake3"
bolt "go.etcd.io/bbolt"
"runtime/pprof"
@ -99,10 +96,6 @@ type TranslateStore struct {
Path string
}
func (s *TranslateStore) GetStorePath() string {
return s.Path
}
// NewTranslateStore returns a new instance of TranslateStore.
func NewTranslateStore(index, field string, partitionID, partitionN int) *TranslateStore {
return &TranslateStore{
@ -579,837 +572,3 @@ func findKeyByID(bkt *bolt.Bucket, id uint64) string {
}
return string(boltKey)
}
func (s *TranslateStore) ComputeTranslatorSummaryRows() (sum *pilosa.TranslatorSummary, err error) {
sum = &pilosa.TranslatorSummary{}
hasher := blake3.New()
err = s.db.View(func(tx *bolt.Tx) error {
bkt := tx.Bucket(bucketKeys)
if bkt == nil {
panic("bucketKeys not found")
}
cur := bkt.Cursor()
for k, v := cur.First(); k != nil; k, v = cur.Next() {
input := append(k, v...)
_, _ = hasher.Write(input)
sum.KeyCount++
}
bkt = tx.Bucket(bucketIDs)
if bkt == nil {
panic("bucketIDs not found")
}
cur = bkt.Cursor()
for k, v := cur.First(); k != nil; k, v = cur.Next() {
input := append(k, v...)
_, _ = hasher.Write(input)
sum.IDCount++
}
return nil
})
if err != nil {
return nil, err
}
var buf [16]byte
_, _ = hasher.Digest().Read(buf[0:])
sum.Checksum = string(buf[:])
return sum, nil
}
func (s *TranslateStore) ComputeTranslatorSummaryCols(partitionID int, topo *pilosa.Topology) (sum *pilosa.TranslatorSummary, err error) {
sum = &pilosa.TranslatorSummary{}
hasher := blake3.New()
if partitionID != s.partitionID {
panic(fmt.Sprintf("inconsistent partitionID arg %v with TranslateStore.paritionID %v", partitionID, s.partitionID))
}
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(topo, topo.Hasher, topo.ReplicaN)
firstPrimary := snap.PrimaryNodeIndex(partitionID)
err = s.db.View(func(tx *bolt.Tx) error {
bkt := tx.Bucket(bucketKeys) // key -> id
if bkt == nil {
panic("bucketKeys not found")
}
cur := bkt.Cursor()
for k, v := cur.First(); k != nil; k, v = cur.Next() {
input := append(k, v...)
//vv("55555 ComputeTranslatorSummaryCols(partitionID=%v, path='%v'), k='%v', v=%x", partitionID, s.Path, string(k), v)
_, _ = hasher.Write(input)
sum.KeyCount++
}
bkt = tx.Bucket(bucketIDs) // id -> key
if bkt == nil {
panic("bucketIDs not found")
}
cur = bkt.Cursor()
for k, v := cur.First(); k != nil; k, v = cur.Next() {
// should the primary be the same for each key in this partition?
id := btou64(k)
shard := id / pilosa.ShardWidth
ks := string(v)
primary := snap.PrimaryForColKeyTranslation(s.index, ks)
if firstPrimary < 0 {
firstPrimary = primary
} else {
if primary != firstPrimary {
panic(fmt.Sprintf("s.index='%v' primary (%v) != firstPrimary (%v); key='%v', id=%v, shard=%v; partitionID=%v; topo='%v'", s.index, primary, firstPrimary, ks, id, shard, partitionID, topo.String()))
}
}
// Verify the invariant that the primaries agree. Just a sanity check.
primaryForShard := snap.PrimaryForShardReplication(s.index, shard)
if primaryForShard != firstPrimary {
panic(fmt.Sprintf("primaryForShard (%v) != firstPrimary (%v); key='%v', id=%v, shard=%v; partitionID=%v", primaryForShard, firstPrimary, ks, id, shard, partitionID))
}
input := append(k, v...)
//vv("55555 ComputeTranslatorSummaryCols(partitionID=%v, path='%v'), idBucket id=%x key='%v'", partitionID, s.Path, id, ks)
_, _ = hasher.Write(input)
sum.IDCount++
}
return nil
})
if err != nil {
return nil, err
}
sum.PrimaryNodeIndex = firstPrimary
var buf [16]byte
_, _ = hasher.Digest().Read(buf[0:])
sum.Checksum = string(buf[:])
return sum, nil
}
func (s *TranslateStore) KeyWalker(walk func(key string, col uint64)) error {
return s.db.View(func(tx *bolt.Tx) error {
bkt := tx.Bucket(bucketKeys)
if bkt == nil {
panic("bucketKeys not found")
}
cur := bkt.Cursor()
for k, v := cur.First(); k != nil; k, v = cur.Next() {
walk(string(k), btou64(v))
}
return nil
})
}
func (s *TranslateStore) IDWalker(walk func(key string, col uint64)) error {
return s.db.View(func(tx *bolt.Tx) error {
bkt := tx.Bucket(bucketIDs)
if bkt == nil {
panic("bucketIDs not found")
}
cur := bkt.Cursor()
for k, v := cur.First(); k != nil; k, v = cur.Next() {
walk(string(v), btou64(k))
}
return nil
})
}
// call s.notifyWrite() when done
func (s *TranslateStore) SetFwdRevMaps(tx *bolt.Tx, fwd map[string]uint64, rev map[uint64]string) (err error) {
localTx := false
if tx == nil {
localTx = true
tx, err = s.db.Begin(true)
if err != nil {
return err
}
defer func() {
_ = tx.Rollback()
}()
}
// reinitialize buckets
err = tx.DeleteBucket(bucketKeys)
if err != nil {
return err
}
err = tx.DeleteBucket(bucketIDs)
if err != nil {
return err
}
if _, err := tx.CreateBucketIfNotExists(bucketKeys); err != nil {
return err
} else if _, err := tx.CreateBucketIfNotExists(bucketIDs); err != nil {
return err
}
key2id := tx.Bucket(bucketKeys)
for k, v := range fwd {
err := key2id.Put([]byte(k), u64tob(v))
if err != nil {
return err
}
}
id2key := tx.Bucket(bucketIDs)
for k, v := range rev {
err := id2key.Put(u64tob(k), []byte(v))
if err != nil {
return err
}
}
if localTx {
return tx.Commit()
}
return nil
}
func (s *TranslateStore) GetFwdRevMaps(tx *bolt.Tx) (fwd map[string]uint64, rev map[uint64]string, err error) {
fwd = make(map[string]uint64)
rev = make(map[uint64]string)
key2id := tx.Bucket(bucketKeys)
err = key2id.ForEach(func(k, v []byte) error {
fwd[string(k)] = btou64(v)
return nil
})
if err != nil {
return
}
id2key := tx.Bucket(bucketIDs)
err = id2key.ForEach(func(k, v []byte) error {
rev[btou64(k)] = string(v)
return nil
})
return
}
//var vv = pilosa.VV
// helpers for repair
// muint64 holds multiple unit64
type muint64 struct {
slc []uint64
}
func (m *muint64) String() (s string) {
for _, e := range m.slc {
s += fmt.Sprintf("%x, ", e)
}
return
}
// mstring holds multiple strings
type mstring struct {
slc []string
}
func (m *mstring) String() (s string) {
for _, e := range m.slc {
s += e + ","
}
return
}
func addToProblemKeys(problemKeys map[string]*muint64, k string, v uint64, noValue bool) {
mu, already := problemKeys[k]
if !already {
mu = &muint64{}
problemKeys[k] = mu
}
if !noValue {
mu.slc = append(mu.slc, v)
}
}
func addToProblemIDs(problemIDs map[uint64]*mstring, k uint64, v string, noValue bool) {
mu, already := problemIDs[k]
if !already {
mu = &mstring{}
problemIDs[k] = mu
}
if !noValue {
mu.slc = append(mu.slc, v)
}
}
// only actually apply the fixes if applyKeyRepairs is true.
// if anything changed, return changed == true.
func (s *TranslateStore) RepairKeys(topo *pilosa.Topology, verbose, applyKeyRepairs bool) (changed bool, err error) {
// strategy: get the full set of keys; the domain keys from
// the forward key->id mapping, and the range keys from the reverse id->key mapping.
// Then march through them and make sure they are mapped correctly.
// At the moment we do try to reuse dangling IDs instead of making
// new ones. This might not always be possible, but we hope for
// now that it suffices b/c it minimizes the amount of fragment
// re-write we may have to do.
/*
// ============ profiling ===============
fd, err := ioutil.TempFile(".", "cpu.prof")
if err != nil {
panic(err)
}
_ = pprof.StartCPUProfile(fd)
defer func() {
pprof.StopCPUProfile()
fd.Close()
}()
// ============ end profiling ===============
*/
tx, err := s.db.Begin(true)
if err != nil {
return false, err
}
defer func() {
_ = tx.Rollback()
}()
fwd, rev, err := s.GetFwdRevMaps(tx)
if err != nil {
return false, err
}
// place to store the correct stuff.
//
// fwd2, rev2: new, repaired versions.
// INVAR: they only contain (correct) invertible mappings.
fwd2 := make(map[string]uint64)
rev2 := make(map[uint64]string)
// and a place to store the problems.
problemKeys := make(map[string]*muint64)
problemIDs := make(map[uint64]*mstring)
fwdscan:
for k, v := range fwd {
_, already := fwd2[k]
if already {
// k has already been repaired. don't worry about further.
continue fwdscan
} else {
// if its already invertible, then just keep it, no need to repair it
// INVAR: k is not in fwd2 (at least not yet).
rkey, ok := rev[v]
if !ok {
// k -> v -> X
addToProblemIDs(problemIDs, v, k, false)
addToProblemKeys(problemKeys, k, v, false)
continue fwdscan
}
if rkey == k {
// yay. a good, invertible, mapping. no repair needed.
if k == "" {
panic("bad empty key")
}
fwd2[k] = v
rev2[v] = k
continue fwdscan
}
// some kind of problem.
// what kind?
// Define problemKey as: 2nd key mapping to id already in fwd2.
// Define problemID as: 2nd ID mapping to key already in fwd2.
// k -> v -> rkey, and rkey != k.
v2, ok := fwd[rkey]
if ok {
// k -> v -> rkey -> v2, where v2 ?= v
if v2 == v {
// k -> v -> rkey -> v
// just have a problemKey in k.
if rkey == "" {
panic("bad empty rkey")
}
fwd2[rkey] = v
rev2[v] = rkey
addToProblemKeys(problemKeys, k, v, true)
continue fwdscan
}
// this is i = 1 test case. :)
// k -> v -> rkey -> v2, where v != v2, and k != rkey.
addToProblemKeys(problemKeys, k, v, false)
addToProblemIDs(problemIDs, v, rkey, false)
continue fwdscan
} else {
// k -> v -> rkey -> X(nil), and rkey != k.
addToProblemKeys(problemKeys, k, v, false)
addToProblemKeys(problemKeys, rkey, 0, true)
addToProblemIDs(problemIDs, v, rkey, false)
}
}
}
revscan:
for id, key := range rev {
k1, already := rev2[id]
_ = k1
if already {
// fine, already there.
continue
}
// if its already invertible, keep it.
rid, ok := fwd[key]
if !ok {
// id -> key -> X
addToProblemKeys(problemKeys, key, 0, true)
addToProblemIDs(problemIDs, id, key, false)
continue revscan
}
if rid == id {
// id -> key -> id. good. but should have been added to fwd2/rev2 above.
panic("should have been added to fwd2/rev2 above!")
} else {
// id -> key -> rid, where id != rid
// so rid -> ?
keyr, ok := rev[rid]
if !ok {
// id -> key -> rid -> X, where id != rid
addToProblemKeys(problemKeys, key, rid, false)
addToProblemKeys(problemKeys, key, id, false)
addToProblemIDs(problemIDs, rid, key, false)
continue revscan
}
if keyr == key {
// id -> key -> rid -> key. So rid is correct and id is dangling.
//
// Heuristic: ASSUME here, that the 2 consistent links key->rid->key are correct,
// and that the single id -> key is in the wrong. This DOESN'T HAVE
// TO BE THE CASE.
if key == "" {
panic("bad empty key")
}
rev2[rid] = key
fwd2[key] = rid
addToProblemIDs(problemIDs, id, "", true)
} else {
// this is test case i = 0. Must handle it.
// id -> key -> rid -> keyr, id != rid, keyr != key.
id2, ok := fwd[keyr]
if ok && id2 == rid {
// id -> key -> rid -> keyr -> rid, id != rid, keyr != key.
// so rid -> keyr -> rid is good.
if keyr == "" {
panic("bad empty keyr")
}
fwd2[keyr] = rid
rev2[rid] = keyr
// and id -> key -> rid is bad, b/c id != rid.
addToProblemKeys(problemKeys, key, rid, false)
addToProblemIDs(problemIDs, id, key, false)
continue revscan
}
// one of these 3 cases holds. all have the same treatment.
// 1) id2 == id: id -> key -> rid -> keyr -> id2; id != rid, rid != id2, keyr != key.
// 2) id2 != id: id -> key -> rid -> keyr -> id2; id != rid, rid != id2, id2 != id, keyr != key.
// 3) !ok: id -> key -> rid -> keyr -> X, id != rid, keyr != key.
addToProblemIDs(problemIDs, id, key, false)
addToProblemKeys(problemKeys, key, rid, false)
addToProblemIDs(problemIDs, rid, keyr, false)
}
}
}
//vv("problemKeys = '%v'", problemKeys)
//vv("problemIDs = '%v'", problemIDs)
newIDs := make(map[uint64]bool)
// assign new IDs to any problemKeys; but first
// try to reuse already allocated IDs that are just dangling.
loopProblemKeys:
for key, ids := range problemKeys {
// first try a minor repair, maybe it was just mssing from rev
// and we can avoid allocate another id.
// sanity check
v2, already := fwd2[key]
if already {
panic(fmt.Sprintf("should not get here since fwd2 is only correct invertibles: key='%v', v2='%x'", key, v2))
}
// INVAR: we have no correct mapping for key in fwd2.
// treat the danglers as "suggestions" for the correction.
for k, id := range ids.slc {
_ = k
_, already = rev2[id]
if !already {
// is this correct?
// id is not in rev2, and key is not in fwd2.
// therefore, we can add them both and maintain consistency.
//vv("add %v to fwd2", key)
if key == "" {
panic("bad empty key")
}
fwd2[key] = id
rev2[id] = key
continue loopProblemKeys
}
}
// INVAR: key -> ? don't know. We didn't find a usable suggestion for the id.
// yes, we get here. We have key. We are looking for a suitable id for it.
// can we get a usable id from the problemIDs?
found := false
suggestions:
for idp, mkeyp := range problemIDs {
for _, candk := range mkeyp.slc {
//vv("checking problemIDs, ipd=%x, candk='%v'; candk==key is %v", idp, candk, candk == key)
if candk == key {
// we have a suggestion from problemIDs that idp might work, doing key -> idp.
// Validate that this is possible.
k2, already := rev2[idp]
_ = k2
if already {
//vv("idp is already in rev2: idp=%v, k2=%v", idp, k2)
continue suggestions
}
// idp works. put it in the correct set.
if key == "" {
panic("bad empty key")
}
rev2[idp] = key
fwd2[key] = idp
found = true
break suggestions
}
}
}
if !found {
id2 := pilosa.GenerateNextPartitionedID(s.index, maxID(tx), s.partitionID, s.partitionN)
//vv("could not minor repair, allocating new id2 = %v instead", id2)
newIDs[id2] = true
if key == "" {
panic("bad empty key")
}
fwd2[key] = id2
rev2[id2] = key
}
} // end problemKeys
//for id, keys := range problemIDs {
//}
if verbose {
reportIfGainedOrLostIDs(s, fwd, fwd2, rev, rev2, newIDs)
reportIfGainedOrLostKeys(s, fwd, fwd2, rev, rev2)
}
adds, changes, changeIDs, err := makeStringKeyChanges(verbose, applyKeyRepairs, tx, s, topo, fwd, fwd2, rev, rev2, newIDs)
if err != nil {
return false, err
}
_, _, _ = adds, changes, changeIDs
//vv("changedIDs = '%#v'", changeIDs)
if len(adds) > 0 || len(changes) > 0 || len(changeIDs) > 0 || len(newIDs) > 0 {
changed = true
}
//vv("newIDs = '%#v'", newIDs)
//vv("fwd2 = '%#v'", fwd2)
//vv("rev2 = '%#v'", rev2)
err = tx.Commit()
if err == nil {
s.notifyWrite()
}
return changed, err
}
func reportIfGainedOrLostIDs(s *TranslateStore, fwd, fwd2 map[string]uint64, rev, rev2 map[uint64]string, newIDs map[uint64]bool) {
// get all IDs ever mentioned
before := make(map[uint64]bool)
after := make(map[uint64]bool)
for _, id := range fwd {
before[id] = true
}
for _, id := range fwd2 {
if !newIDs[id] {
after[id] = true
}
}
for id := range rev {
before[id] = true
}
for id := range rev2 {
if !newIDs[id] {
after[id] = true
}
}
nb := len(before)
na := len(after)
if nb != na {
fmt.Printf("# needs-repair: Num ID before %v != Num ID after %v, for boltdb = '%v'. before counts(fwd/rev) = %v/%v. after repair counts(fwd2/rev2) = %v/%v\n", nb, na, s.Path, len(fwd), len(rev), len(fwd2), len(rev2))
}
if len(newIDs) > 0 {
fmt.Printf("# needs-repair: adding newIDs '%#v', for boltdb = '%v'. before counts(fwd/rev) = %v/%v. after repair counts(fwd2/rev2) = %v/%v\n", newIDs, s.Path, len(fwd), len(rev), len(fwd2), len(rev2))
}
}
func reportIfGainedOrLostKeys(s *TranslateStore, fwd, fwd2 map[string]uint64, rev, rev2 map[uint64]string) {
// get all IDs ever mentioned
nb := len(fwd)
na := len(fwd2)
if nb != na {
diffAB := mapDiffStrings(fwd, fwd2)
diffBA := mapDiffStrings(fwd2, fwd)
fmt.Printf("# needs-repair: Num Keys before != Num Keys after, for boltdb = '%v'. before counts(fwd/rev) = %v/%v. after repair counts(fwd2/rev2) = %v/%v. fwd - fwd2 = '%#v'; fwd2-fwd = '%#v'\n", s.Path, len(fwd), len(rev), len(fwd2), len(rev2), diffAB, diffBA)
}
}
// return A - B
func mapDiffStrings(mapA, mapB map[string]uint64) (r []string) {
for a := range mapA {
_, ok := mapB[a]
if !ok {
r = append(r, a)
}
}
sort.Strings(r)
return
}
type BeforeAfterKeyChange struct {
BeforeID uint64
AfterID uint64
}
type BeforeAfterIDChange struct {
IsDelete bool
IsAdd bool
BeforeString string
AfterString string
}
// do the minimal state update.
// fwd2 is the "after" map, all string keys repaired.
func makeStringKeyChanges(
verbose bool,
applyKeyRepairs bool,
tx *bolt.Tx,
s *TranslateStore,
topo *pilosa.Topology,
fwd, fwd2 map[string]uint64,
rev, rev2 map[uint64]string,
newIDs map[uint64]bool,
) (
adds map[string]uint64,
changeKeys map[string]*BeforeAfterKeyChange,
changeIDs map[uint64]*BeforeAfterIDChange,
err error,
) {
//vv("makeStringKeyChanges called")
//vv("fwd2 = '%#v'", fwd2)
//vv("rev2 = '%#v'", rev2)
//vv("fwd = '%#v'", fwd)
//vv("rev = '%#v'", rev)
var action string
if applyKeyRepairs {
action = "applying "
}
// addition of string key
adds = make(map[string]uint64)
// change of the mapping of key -> id.
changeKeys = make(map[string]*BeforeAfterKeyChange)
// changes to bucketIDs
changeIDs = make(map[uint64]*BeforeAfterIDChange)
localTx := false
if applyKeyRepairs && tx == nil {
localTx = true
tx, err = s.db.Begin(true)
if err != nil {
return
}
defer func() {
//vv("tx.Rollback happening")
_ = tx.Rollback()
}()
}
key2id := tx.Bucket(bucketKeys)
id2key := tx.Bucket(bucketIDs)
// make a copy of rev2 that we can delete from, to see if
// any additions left in rev2 need to be added after all of
// rev is analyzed.
rev2cp := make(map[uint64]string)
for id, k := range rev2 {
rev2cp[id] = k
}
// first we clean up any stale IDs from id2key. Then the fwd2 pass
// that follows will write to both key2id and id2key.
for id, key := range rev {
//vv("makeStringKeyChanges on rev2: id=%x -> key='%v'", id, key)
key2, ok := rev2[id]
if !ok {
changeIDs[id] = &BeforeAfterIDChange{IsDelete: true}
if verbose {
fmt.Printf("# %vkey-translation-delete-id: (id %x -> %v). Remaining for that key: ('%v' -> %x)\n", action, id, key, key, fwd2[key])
}
if applyKeyRepairs {
u := u64tob(id)
err = id2key.Delete(u)
if err != nil {
return
}
}
continue
}
delete(rev2cp, id)
if key2 != key {
u := u64tob(id)
k := []byte(key2)
changeIDs[id] = &BeforeAfterIDChange{
BeforeString: key,
AfterString: key2,
}
if verbose {
fmt.Printf("# %vkey-translation-update-id: (id %x -> %v). fwd2 for that key: ('%v' -> %x)\n", action, id, key2, key2, fwd2[key2])
}
if applyKeyRepairs {
err = id2key.Put(u, k)
if err != nil {
return
}
}
}
}
// anything leftover in rev2cp is stuff that is new, only
// in rev2 and not in rev. It needs to be added.
for id, key2 := range rev2cp {
u := u64tob(id)
k := []byte(key2)
changeIDs[id] = &BeforeAfterIDChange{
IsAdd: true,
//BeforeString: left empty
AfterString: key2,
}
if verbose {
fmt.Printf("# %vkey-translation-add-id: (id %x -> %v). Fwd for that key: ('%v' -> %x)\n", action, id, key2, key2, fwd2[key2])
}
if applyKeyRepairs {
err = id2key.Put(u, k)
if err != nil {
return
}
}
}
// We assume here that fwd2 is a super-set of fwd. No string keys
// should be deleted in the repair. Confirm that.
for key, id := range fwd {
_, ok := fwd2[key]
if !ok {
panic(fmt.Sprintf("fwd2 is missing a string key from fwd. key='%v' -> id='%x'", key, id))
}
}
// Create a snapshot of the cluster to use for node/partition calculations.
var snap *topology.ClusterSnapshot
if topo != nil {
snap = topology.NewClusterSnapshot(topo, topo.Hasher, topo.ReplicaN)
}
for key2, id2 := range fwd2 {
//vv("makeStringKeyChanges on fwd2, key2='%v', id2=%x", key2, id2)
isPrimary := false
if topo != nil {
primary := snap.PrimaryForColKeyTranslation(s.index, key2)
isPrimary = s.partitionID == primary
}
_ = isPrimary
id, ok := fwd[key2]
if !ok {
adds[key2] = id2
u2 := u64tob(id2)
k2 := []byte(key2)
if verbose {
fmt.Printf("# %vkey-translation-new-key: ('%v' -> %x) added: isPrimary: %v\n", action, key2, id2, isPrimary)
}
if applyKeyRepairs {
err = key2id.Put(k2, u2)
if err != nil {
return
}
err = id2key.Put(u2, k2)
if err != nil {
return
}
}
continue
}
if id != id2 {
changeKeys[key2] = &BeforeAfterKeyChange{
BeforeID: id,
AfterID: id2,
}
if verbose {
fmt.Printf("# %vkey-translation-change-id: ('%v' -> %x) changes to ('%v' -> %x); isPrimary: %v\n", action, key2, id, key2, id2, isPrimary)
}
if applyKeyRepairs {
u2 := u64tob(id2)
k2 := []byte(key2)
err = key2id.Put(k2, u2)
if err != nil {
return
}
err = id2key.Put(u2, k2)
if err != nil {
return
}
}
}
}
if localTx {
err = tx.Commit()
}
return
}
func (s *TranslateStore) DumpBolt(label string) {
fmt.Printf("dumping bolt %v : path='%v'\n", label, s.Path)
_ = s.KeyWalker(func(key string, col uint64) {
fmt.Printf("keyWalker: key '%v' -> col '%x'\n", key, col)
})
_ = s.IDWalker(func(key string, col uint64) {
fmt.Printf("idWalker: id '%x' -> key '%v'\n", col, key)
})
fmt.Printf("DONE with dumping bolt %v; path='%v'\n", label, s.Path)
}

170
boltdb/translate_test.go
View file

@ -628,173 +628,3 @@ func MustCloseTranslateStore(s *boltdb.TranslateStore) {
panic(err)
}
}
func TestCryptoHashPerKey(t *testing.T) {
s := MustOpenNewTranslateStore()
defer MustCloseTranslateStore(s)
// hash one translation
expect := map[int]string{
1: string([]byte{0x76, 0x48, 0x8b, 0x70, 0xe8, 0x54, 0x35, 0xc6, 0x8e, 0xa6, 0x4, 0x6c, 0xfa, 0xd2, 0x1a, 0x12}),
2: string([]byte{0x81, 0x46, 0x84, 0x37, 0x26, 0x96, 0x41, 0xf3, 0x54, 0x4e, 0x98, 0xbc, 0x48, 0xab, 0x1b, 0xf0}),
3: string([]byte{0x7f, 0xe9, 0xf, 0x6d, 0x7b, 0x14, 0x1, 0x44, 0xb2, 0x4e, 0xd0, 0x86, 0x2f, 0x62, 0x8c, 0xa9}),
}
for n := 1; n < 4; n++ {
var batch0 []string
for i := 0; i < n; i++ {
batch0 = append(batch0, fmt.Sprintf("key%d", i))
}
// Populate the store with the keys in batch0.
batch0IDs, err := s.TranslateKeys(batch0, true)
_ = batch0IDs
if err != nil {
t.Fatal(err)
}
// done with setup
sum, err := s.ComputeTranslatorSummaryCols(0, pilosa.NewTopology(&topology.Jmphasher{}, topology.DefaultPartitionN, 1, nil))
if err != nil {
panic(err)
}
nkey := sum.KeyCount
nid := sum.IDCount
observedChecksum := sum.Checksum
if nkey != n {
panic("wrong key count")
}
if nkey != nid {
panic("key count should match id count")
}
// shardwidth 22 has different hashes, of course.
if pilosa.ShardWidth == 20 {
expectedChecksum := expect[n]
if observedChecksum != expectedChecksum {
panic(fmt.Sprintf("got wrong checksum obs '%#v' vs expected '%#v'", observedChecksum, expectedChecksum))
}
}
}
}
func TestTranslateStore_RepairNonInvertibleStringKeyTranslation(t *testing.T) {
const N = 6
// before repair
var fwd [N]map[string]uint64
var rev [N]map[uint64]string
// after repair
var fwd2 [N]map[string]uint64
var rev2 [N]map[uint64]string
// case 0: forward is messed up (unlikely but check for it anyway, be sure we can repair)
// "key0" -> id 0 // correct.
// "key1" -> id 0 // wrong. after Repair, should see key1 -> 1 (0xec0002)
//
// id 0 -> "key0" // correct
// id 1 -> "key1" // correct
//
fwd[0] = map[string]uint64{"key0": 0xec00001, "key1": 0xec00001}
rev[0] = map[uint64]string{0xec00001: "key0", 0xec00002: "key1"}
fwd2[0] = map[string]uint64{"key0": 0xec00001, "key1": 0xec00002}
rev2[0] = map[uint64]string{0xec00001: "key0", 0xec00002: "key1"}
// case 1: reverse is messed up (we have seen this in the past)
// "key0" -> id 0 // correct
// "key1" -> id 1 // correct
//
// id 0 -> "key0" // correct.
// id 1 -> "key0" // wrong. after Repair, should see id 1 -> "key1"
//
fwd[1] = map[string]uint64{"key0": 0xec00001, "key1": 0xec00002}
rev[1] = map[uint64]string{0xec00001: "key0", 0xec00002: "key0"}
fwd2[1] = map[string]uint64{"key0": 0xec00001, "key1": 0xec00002}
rev2[1] = map[uint64]string{0xec00001: "key0", 0xec00002: "key1"}
// case 2: only present in reverse.
fwd[2] = map[string]uint64{}
rev[2] = map[uint64]string{0xec00001: "key0"}
fwd2[2] = map[string]uint64{"key0": 0xec00001}
rev2[2] = map[uint64]string{0xec00001: "key0"}
// case 3: same thing. with camoflage.
fwd[3] = map[string]uint64{"key1": 0xec00002}
rev[3] = map[uint64]string{0xec00001: "key0", 0xec00002: "key1"}
fwd2[3] = map[string]uint64{"key0": 0xec00001, "key1": 0xec00002}
rev2[3] = map[uint64]string{0xec00001: "key0", 0xec00002: "key1"}
// case 4: only present in forward.
fwd[4] = map[string]uint64{"key0": 0xec00001}
rev[4] = map[uint64]string{}
fwd2[4] = map[string]uint64{"key0": 0xec00001}
rev2[4] = map[uint64]string{0xec00001: "key0"}
// case 5: same thing. with camoflage.
fwd[5] = map[string]uint64{"key0": 0xec00001}
rev[5] = map[uint64]string{0xec00002: "key1"}
fwd2[5] = map[string]uint64{"key0": 0xec00001, "key1": 0xec00002}
rev2[5] = map[uint64]string{0xec00001: "key0", 0xec00002: "key1"}
// case 6: we had an id, but b/c of the fix, that id is no longer used.
// now that id might still be used in the fragment for a column,
// and so we will need to remove that id/column from the fragment.
// encapsulated: "did it affect the state of the fields?"
for i := 0; i < 5; i++ {
//println("i = ", i)
s := MustOpenNewTranslateStore()
defer MustCloseTranslateStore(s)
if err := s.SetFwdRevMaps(nil, fwd[i], rev[i]); err != nil {
t.Fatal(err)
}
if err := verifyState("setup", i, s, fwd[i], rev[i]); err != nil {
t.Fatal(err)
}
var topo *pilosa.Topology
verbose := false
applyKeyRepairs := true
changed, err := s.RepairKeys(topo, verbose, applyKeyRepairs)
if err != nil {
t.Fatal(err)
}
if !changed {
t.Fatalf("expected changes!")
}
if err := verifyState("afterRepair", i, s, fwd2[i], rev2[i]); err != nil {
t.Fatal(err)
}
}
}
func verifyState(label string, i int, s *boltdb.TranslateStore, fwd map[string]uint64, rev map[uint64]string) error {
// verify the setup
const writable = true
for key, expectID := range fwd {
id, err := s.TranslateKey(key, !writable)
if err != nil {
return err
}
if id != expectID {
return fmt.Errorf("fwd %v problem. i=%v, for key '%v', expected %x, observed %x", label, i, key, expectID, id)
}
}
for id, expectKey := range rev {
key, err := s.TranslateID(id)
if err != nil {
return err
}
if key != expectKey {
return fmt.Errorf("rev %v problem. i=%v, for id '%x', expected %v, observed %v", label, i, id, expectKey, key)
}
}
return nil
}

22
mock/translator.go
View file

@ -37,13 +37,6 @@ type TranslateStore struct {
EntryReaderFunc func(ctx context.Context, offset uint64) (pilosa.TranslateEntryReader, error)
}
func (s *TranslateStore) ComputeTranslatorSummaryRows() (sum *pilosa.TranslatorSummary, err error) {
return
}
func (s *TranslateStore) ComputeTranslatorSummaryCols(partitionID int, topo *pilosa.Topology) (sum *pilosa.TranslatorSummary, err error) {
return
}
func (s *TranslateStore) Close() error {
return s.CloseFunc()
}
@ -104,14 +97,6 @@ func (s *TranslateStore) ReadFrom(r io.Reader) (int64, error) {
return 0, nil
}
func (s *TranslateStore) RepairKeys(topo *pilosa.Topology, verbose, applyKeyRepairs bool) (changed bool, err error) {
return
}
func (s *TranslateStore) GetStorePath() string {
return ""
}
var _ pilosa.TranslateEntryReader = (*TranslateEntryReader)(nil)
type TranslateEntryReader struct {
@ -126,10 +111,3 @@ func (r *TranslateEntryReader) Close() error {
func (r *TranslateEntryReader) ReadEntry(entry *pilosa.TranslateEntry) error {
return r.ReadEntryFunc(entry)
}
func (s *TranslateStore) KeyWalker(walk func(key string, col uint64)) error {
panic("TODO")
}
func (s *TranslateStore) IDWalker(walk func(key string, col uint64)) error {
panic("TODO")
}

46
translate.go
View file

@ -99,16 +99,6 @@ type TranslateStore interface { // TODO: refactor this interface; readonly shoul
// It should read from the reader and replace the data store with
// the read payload.
ReadFrom(io.Reader) (int64, error)
ComputeTranslatorSummaryRows() (sum *TranslatorSummary, err error)
ComputeTranslatorSummaryCols(partitionID int, topo *Topology) (sum *TranslatorSummary, err error)
KeyWalker(walk func(key string, col uint64)) error
IDWalker(walk func(key string, col uint64)) error
RepairKeys(topo *Topology, verbose, applyKeyRepairs bool) (changed bool, err error)
GetStorePath() string
}
// TranslatorSummary is returned, for example from the boltdb string key translators,
@ -365,30 +355,6 @@ func NewInMemTranslateStore(index, field string, partitionID, partitionN int) *I
}
}
func (s *InMemTranslateStore) GetStorePath() string {
return ""
}
// KeyWalker executes walk for every pair in the database
func (s *InMemTranslateStore) KeyWalker(walk func(key string, col uint64)) error {
s.mu.RLock()
defer s.mu.RUnlock()
for id, key := range s.keysByID {
walk(key, id)
}
return nil
}
// IDWalker executes walk for every pair in the database
func (s *InMemTranslateStore) IDWalker(walk func(key string, col uint64)) error {
s.mu.RLock()
defer s.mu.RUnlock()
for key, id := range s.idsByKey {
walk(key, id)
}
return nil
}
var _ OpenTranslateStoreFunc = OpenInMemTranslateStore
// OpenInMemTranslateStore returns a new instance of InMemTranslateStore.
@ -397,18 +363,6 @@ func OpenInMemTranslateStore(rawurl, index, field string, partitionID, partition
return NewInMemTranslateStore(index, field, partitionID, partitionN), nil
}
func (s *InMemTranslateStore) ComputeTranslatorSummaryRows() (sum *TranslatorSummary, err error) {
panic("TODO")
}
func (s *InMemTranslateStore) ComputeTranslatorSummaryCols(partitionID int, topo *Topology) (sum *TranslatorSummary, err error) {
panic("TODO")
}
func (s *InMemTranslateStore) RepairKeys(topo *Topology, verbose, applyKeyRepairs bool) (changed bool, err error) {
panic("TODO")
}
func (s *InMemTranslateStore) Close() error {
return nil
}