// 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 boltdb import ( "bytes" "context" "fmt" "io" "io/ioutil" "os" "path/filepath" "sort" "sync" "time" "github.com/pilosa/pilosa/v2" "github.com/pkg/errors" "github.com/zeebo/blake3" bolt "go.etcd.io/bbolt" "runtime/pprof" ) var _ = ioutil.TempFile var _ = pprof.StartCPUProfile var ( // ErrTranslateStoreClosed is returned when reading from an TranslateEntryReader // and the underlying store is closed. ErrTranslateStoreClosed = errors.New("boltdb: translate store closing") // ErrTranslateKeyNotFound is returned when translating key // and the underlying store returns an empty set ErrTranslateKeyNotFound = errors.New("boltdb: translating key returned empty set") bucketKeys = []byte("keys") bucketIDs = []byte("ids") ) const ( // snapshotExt is the file extension used for an in-process snapshot. snapshotExt = ".snapshotting" errFmtTranslateBucketNotFound = "boltdb: translate bucket '%s' not found" ) // OpenTranslateStore opens and initializes a boltdb translation store. func OpenTranslateStore(path, index, field string, partitionID, partitionN int) (pilosa.TranslateStore, error) { s := NewTranslateStore(index, field, partitionID, partitionN) s.Path = path if err := s.Open(); err != nil { return nil, err } return s, nil } // Ensure type implements interface. var _ pilosa.TranslateStore = &TranslateStore{} // TranslateStore is an on-disk storage engine for translating string-to-uint64 values. // An empty string will be converted into the sentinel byte slice: // var emptyKey = []byte{ // 0x00, 0x00, 0x00, // 0x4d, 0x54, 0x4d, 0x54, // MTMT // 0x00, // 0xc2, 0xa0, // NO-BREAK SPACE // 0x00, // } type TranslateStore struct { mu sync.RWMutex db *bolt.DB index string field string partitionID int partitionN int once sync.Once closing chan struct{} readOnly bool writeNotify chan struct{} // File path to database file. 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{ index: index, field: field, partitionID: partitionID, partitionN: partitionN, closing: make(chan struct{}), writeNotify: make(chan struct{}), } } // Open opens the translate file. func (s *TranslateStore) Open() (err error) { // add the path to the problem database if we panic handling it. defer func() { r := recover() if r != nil { panic(fmt.Sprintf("pilosa/boltdb/TranslateStore.Open(s.Path='%v') panic with '%v'", s.Path, r)) } }() if err := os.MkdirAll(filepath.Dir(s.Path), 0777); err != nil { return errors.Wrapf(err, "mkdir %s", filepath.Dir(s.Path)) } else if s.db, err = bolt.Open(s.Path, 0666, &bolt.Options{Timeout: 1 * time.Second}); err != nil { return errors.Wrapf(err, "open file: %s", err) } // Initialize buckets. if err := s.db.Update(func(tx *bolt.Tx) error { if _, err := tx.CreateBucketIfNotExists(bucketKeys); err != nil { return err } else if _, err := tx.CreateBucketIfNotExists(bucketIDs); err != nil { return err } return nil }); err != nil { s.db.Close() return err } return nil } // Close closes the underlying database. func (s *TranslateStore) Close() (err error) { s.once.Do(func() { close(s.closing) }) if s.db != nil { if err := s.db.Close(); err != nil { return err } } return nil } // PartitionID returns the partition id the store was initialized with. func (s *TranslateStore) PartitionID() int { return s.partitionID } // ReadOnly returns true if the store is in read-only mode. func (s *TranslateStore) ReadOnly() bool { s.mu.RLock() defer s.mu.RUnlock() return s.readOnly } // SetReadOnly toggles whether store is in read-only mode. func (s *TranslateStore) SetReadOnly(v bool) { s.mu.Lock() defer s.mu.Unlock() s.readOnly = v } // Size returns the number of bytes in the data file. func (s *TranslateStore) Size() int64 { if s.db == nil { return 0 } tx, err := s.db.Begin(false) if err != nil { return 0 } defer func() { _ = tx.Rollback() }() return tx.Size() } // TranslateKey converts a string key to an integer ID. // If key does not have an associated id then one is created, unless writable is false, // then the function will return the error pilosa.ErrTranslatingKeyNotFound. func (s *TranslateStore) TranslateKey(key string, writable bool) (uint64, error) { ids, err := s.translateKeys([]string{key}, writable) if err != nil { return 0, err } if len(ids) == 0 { return 0, ErrTranslateKeyNotFound } return ids[0], nil } // TranslateKeys converts a slice of string keys to a slice of integer IDs. // If a key does not have an associated id then one is created, unless writable is false, // then the function will return the error pilosa.ErrTranslatingKeyNotFound. func (s *TranslateStore) TranslateKeys(keys []string, writable bool) ([]uint64, error) { return s.translateKeys(keys, writable) } // FindKeys looks up the ID for each key. // Keys are not created if they do not exist. // Missing keys are not considered errors, so the length of the result may be less than that of the input. func (s *TranslateStore) FindKeys(keys ...string) (map[string]uint64, error) { result := make(map[string]uint64, len(keys)) err := s.db.View(func(tx *bolt.Tx) error { bkt := tx.Bucket(bucketKeys) if bkt == nil { return errors.Errorf(errFmtTranslateBucketNotFound, bucketKeys) } for _, key := range keys { id, _ := findIDByKey(bkt, key) if id == 0 { // The key does not exist. continue } result[key] = id } return nil }) if err != nil { return nil, err } return result, nil } // CreateKeys maps all keys to IDs, creating the IDs if they do not exist. // If the translator is read-only, this will return an error. func (s *TranslateStore) CreateKeys(keys ...string) (map[string]uint64, error) { if s.ReadOnly() { return nil, pilosa.ErrTranslateStoreReadOnly } written := false result := make(map[string]uint64, len(keys)) err := s.db.Update(func(tx *bolt.Tx) error { keyBucket := tx.Bucket(bucketKeys) if keyBucket == nil { return errors.Errorf(errFmtTranslateBucketNotFound, bucketKeys) } idBucket := tx.Bucket(bucketIDs) if keyBucket == nil { return errors.Errorf(errFmtTranslateBucketNotFound, bucketIDs) } for _, key := range keys { id, boltKey := findIDByKey(keyBucket, key) if id == 0 { // The key does not exist. id = pilosa.GenerateNextPartitionedID(s.index, maxID(tx), s.partitionID, s.partitionN) if err := keyBucket.Put(boltKey, u64tob(id)); err != nil { return err } else if err := idBucket.Put(u64tob(id), boltKey); err != nil { return err } written = true } result[key] = id } return nil }) if err != nil { return nil, err } if written { s.notifyWrite() } return result, nil } func (s *TranslateStore) translateKeys(keys []string, writable bool) ([]uint64, error) { ids := make([]uint64, 0, len(keys)) if s.ReadOnly() || !writable { found := 0 if err := s.db.View(func(tx *bolt.Tx) error { bkt := tx.Bucket(bucketKeys) if bkt == nil { return errors.Errorf(errFmtTranslateBucketNotFound, bucketKeys) } for _, key := range keys { if id, _ := findIDByKey(bkt, key); id != 0 { ids = append(ids, id) found++ } } return nil }); err != nil { return nil, err } if found == len(keys) { return ids, nil } if s.ReadOnly() { return ids, pilosa.ErrTranslateStoreReadOnly } if !writable { return nil, pilosa.ErrTranslatingKeyNotFound } return nil, nil } // Find or create ids under write lock if any keys were not found. var written bool if err := s.db.Update(func(tx *bolt.Tx) (err error) { bkt := tx.Bucket(bucketKeys) for _, key := range keys { id, boltKey := findIDByKey(bkt, key) if id != 0 { ids = append(ids, id) continue } id = pilosa.GenerateNextPartitionedID(s.index, maxID(tx), s.partitionID, s.partitionN) if err := bkt.Put(boltKey, u64tob(id)); err != nil { return err } else if err := tx.Bucket(bucketIDs).Put(u64tob(id), boltKey); err != nil { return err } ids = append(ids, id) written = true } return nil }); err != nil { return nil, err } if written { s.notifyWrite() } return ids, nil } // TranslateID converts an integer ID to a string key. // Returns a blank string if ID does not exist. func (s *TranslateStore) TranslateID(id uint64) (string, error) { tx, err := s.db.Begin(false) if err != nil { return "", err } defer func() { _ = tx.Rollback() }() return findKeyByID(tx.Bucket(bucketIDs), id), nil } // TranslateIDs converts a list of integer IDs to a list of string keys. func (s *TranslateStore) TranslateIDs(ids []uint64) ([]string, error) { if len(ids) == 0 { return nil, nil } tx, err := s.db.Begin(false) if err != nil { return nil, err } defer func() { _ = tx.Rollback() }() bucket := tx.Bucket(bucketIDs) keys := make([]string, len(ids)) for i, id := range ids { keys[i] = findKeyByID(bucket, id) } return keys, nil } // ForceSet writes the id/key pair to the store even if read only. Used by replication. func (s *TranslateStore) ForceSet(id uint64, key string) error { if err := s.db.Update(func(tx *bolt.Tx) (err error) { if err := tx.Bucket(bucketKeys).Put([]byte(key), u64tob(id)); err != nil { return err } else if err := tx.Bucket(bucketIDs).Put(u64tob(id), []byte(key)); err != nil { return err } return nil }); err != nil { return err } s.notifyWrite() return nil } // EntryReader returns a reader that streams the underlying data file. func (s *TranslateStore) EntryReader(ctx context.Context, offset uint64) (pilosa.TranslateEntryReader, error) { ctx, cancel := context.WithCancel(ctx) return &TranslateEntryReader{ctx: ctx, cancel: cancel, store: s, offset: offset}, nil } // WriteNotify returns a channel that is closed when a new entry is written. func (s *TranslateStore) WriteNotify() <-chan struct{} { s.mu.RLock() ch := s.writeNotify s.mu.RUnlock() return ch } // notifyWrite sends a write notification under write lock. func (s *TranslateStore) notifyWrite() { s.mu.Lock() defer s.mu.Unlock() close(s.writeNotify) s.writeNotify = make(chan struct{}) } // MaxID returns the highest id in the store. func (s *TranslateStore) MaxID() (max uint64, err error) { if err := s.db.View(func(tx *bolt.Tx) error { max = maxID(tx) return nil }); err != nil { return 0, err } return max, nil } // WriteTo writes the contents of the store to the writer. func (s *TranslateStore) WriteTo(w io.Writer) (int64, error) { tx, err := s.db.Begin(false) if err != nil { return 0, err } defer func() { _ = tx.Rollback() }() return tx.WriteTo(w) } // ReadFrom reads the content and overwrites the existing store. func (s *TranslateStore) ReadFrom(r io.Reader) (n int64, err error) { // Close store. if err := s.Close(); err != nil { return 0, errors.Wrap(err, "closing store") } // Create a temporary file to snapshot to. snapshotPath := s.Path + snapshotExt file, err := os.Create(snapshotPath) if err != nil { return n, errors.Wrap(err, "creating snapshot file") } // Write payload to snapshot. if n, err = io.Copy(file, r); err != nil { file.Close() return n, errors.Wrap(err, "snapshot write to") } // we close the file here so we don't still have it open when trying // to open it in a moment. file.Close() // Move snapshot to data file location. if err := os.Rename(snapshotPath, s.Path); err != nil { return n, errors.Wrap(err, "renaming snapshot") } // Re-open the store. if err := s.Open(); err != nil { return n, errors.Wrap(err, "re-opening store") } return n, nil } // MaxID returns the highest id in the store. func maxID(tx *bolt.Tx) uint64 { if key, _ := tx.Bucket(bucketIDs).Cursor().Last(); key != nil { return btou64(key) } return 0 } type TranslateEntryReader struct { ctx context.Context store *TranslateStore offset uint64 cancel func() } // Close closes the reader. func (r *TranslateEntryReader) Close() error { r.cancel() return nil } // ReadEntry reads the next entry from the underlying translate store. func (r *TranslateEntryReader) ReadEntry(entry *pilosa.TranslateEntry) error { // Ensure reader has not been closed before read. select { case <-r.ctx.Done(): return r.ctx.Err() case <-r.store.closing: return ErrTranslateStoreClosed default: } for { // Obtain notification channel before read to ensure concurrency issues. writeNotify := r.store.WriteNotify() // Find next ID/key pair in transaction. var found bool if err := r.store.db.View(func(tx *bolt.Tx) error { // Find ID/key lookup at offset or later. cur := tx.Bucket(bucketIDs).Cursor() key, value := cur.Seek(u64tob(r.offset)) if key == nil { return nil } // Copy ID & key to entry and mark as found. found = true entry.Index = r.store.index entry.Field = r.store.field entry.ID = btou64(key) entry.Key = string(value) // Update offset position. r.offset = entry.ID + 1 return nil }); err != nil { return err } else if found { return nil } // If no entry found, wait for new write or reader close. select { case <-r.ctx.Done(): return r.ctx.Err() case <-r.store.closing: return ErrTranslateStoreClosed case <-writeNotify: } } } // emptyKey is a sentinel byte slice which stands for "" as a key. var emptyKey = []byte{ 0x00, 0x00, 0x00, 0x4d, 0x54, 0x4d, 0x54, // MTMT 0x00, 0xc2, 0xa0, // NO-BREAK SPACE 0x00, } func findIDByKey(bkt *bolt.Bucket, key string) (uint64, []byte) { var boltKey []byte if key == "" { boltKey = emptyKey } else { boltKey = []byte(key) } if value := bkt.Get(boltKey); value != nil { return btou64(value), boltKey } return 0, boltKey } func findKeyByID(bkt *bolt.Bucket, id uint64) string { boltKey := bkt.Get(u64tob(id)) if bytes.Equal(boltKey, emptyKey) { return "" } 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)) } firstPrimary := topo.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 := topo.GetPrimaryForColKeyTranslation(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 := topo.GetPrimaryForShardReplication(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)) } } for key2, id2 := range fwd2 { //vv("makeStringKeyChanges on fwd2, key2='%v', id2=%x", key2, id2) isPrimary := false if topo != nil { primary := topo.GetPrimaryForColKeyTranslation(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) }