diff --git a/boltdb/translate.go b/boltdb/translate.go index 40ba72cd3..00f0e657c 100644 --- a/boltdb/translate.go +++ b/boltdb/translate.go @@ -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) - -} diff --git a/boltdb/translate_test.go b/boltdb/translate_test.go index 1da4bf0ed..6908da6ac 100644 --- a/boltdb/translate_test.go +++ b/boltdb/translate_test.go @@ -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 -} diff --git a/mock/translator.go b/mock/translator.go index 9be03715a..e7e88644f 100644 --- a/mock/translator.go +++ b/mock/translator.go @@ -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") -} diff --git a/translate.go b/translate.go index e8ed1f083..71bf45b4d 100644 --- a/translate.go +++ b/translate.go @@ -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 }