diff --git a/api.go b/api.go index 773c73ae4..5edab36ee 100644 --- a/api.go +++ b/api.go @@ -1870,6 +1870,7 @@ func (api *API) ActiveQueries(ctx context.Context) ([]ActiveQueryStatus, error) } // TranslateIndexDB is an internal function to load the index keys database +// rd is a boltdb file. func (api *API) TranslateIndexDB(ctx context.Context, indexName string, partitionID int, rd io.Reader) error { idx := api.holder.Index(indexName) store := idx.TranslateStore(partitionID) diff --git a/boltdb/translate.go b/boltdb/translate.go index 4977c7670..55c837eac 100644 --- a/boltdb/translate.go +++ b/boltdb/translate.go @@ -26,6 +26,7 @@ import ( "github.com/boltdb/bolt" "github.com/pilosa/pilosa/v2" "github.com/pkg/errors" + "github.com/zeebo/blake3" ) var ( @@ -480,3 +481,43 @@ func findKeyByID(bkt *bolt.Bucket, id uint64) string { } return string(boltKey) } + +func (s *TranslateStore) ComputeTranslatorSummary() (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 +} diff --git a/boltdb/translate_test.go b/boltdb/translate_test.go index 73b4e5345..8c7a23c18 100644 --- a/boltdb/translate_test.go +++ b/boltdb/translate_test.go @@ -482,3 +482,54 @@ 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.ComputeTranslatorSummary() + 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)) + } + } + } + +} diff --git a/cmd/pilosa-chk/chk.go b/cmd/pilosa-chk/chk.go new file mode 100644 index 000000000..eebf3333b --- /dev/null +++ b/cmd/pilosa-chk/chk.go @@ -0,0 +1,101 @@ +// Copyright 2020 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 main + +import ( + "flag" + "fmt" + "log" + "os" + + "github.com/pilosa/pilosa/v2" + "github.com/pilosa/pilosa/v2/boltdb" + "github.com/zeebo/blake3" +) + +// pilosa-chk : read boltdb files and print checksums and counts on the keys. With +// -v and -ops and -bits you can display every last bit if you want. +// +// pilosa-chk is deliberately NOT a part of pilosa so that it can run without +// forcing a customer to upgrade or downgrade their installed version. + +func main() { + + var dir string + var showOpsLog bool + var showBits bool + var showFrags bool + home := os.Getenv("HOME") + flag.StringVar(&dir, "dir", fmt.Sprintf("%v/.pilosa", home), "pilosa data dir to read") + flag.BoolVar(&showFrags, "v", false, "show the checksum hash for each fragment in each index. Warning: long output") + flag.BoolVar(&showOpsLog, "ops", false, "show the ops log for each fragment. Warning: very long output. Implies -v") + flag.BoolVar(&showBits, "bits", false, "show the hot bits for each fragment. Warning: very, very long output. Implies -v") + flag.Parse() + + if showBits { + showFrags = true + } + if showOpsLog { + showFrags = true + } + fmt.Printf("opening dir '%v'... this may take a few seconds...\n", dir) + fmt.Printf(" the blake-3 hash includes the value of each mapping and the field or partitionID.\n") + + holder := pilosa.NewHolder(256) + holder.Path = dir + holder.OpenTranslateStore = boltdb.OpenTranslateStore + + err := holder.Open() + + if err != nil { + log.Fatal(err) + } + + fmt.Printf("\ncalculating hashes of row and column key translation maps on data from dir '%v'...\n", dir) + var indexes []*pilosa.Index + + final := pilosa.NewAllTranslatorSummary() + const verbose = true + for _, idx := range holder.Indexes() { + asum, err := idx.ComputeTranslatorSummary(verbose) + if err != nil { + log.Fatal(err) + } + final.Append(asum) + indexes = append(indexes, idx) + } + final.Sort() + + hasher := blake3.New() + fmt.Printf("\nsummary of %v:\n", dir) + for _, sum := range final.Sums { + //fmt.Printf("index: %v partitionID: %v blake3-%x keyCount: %v idCount: %v\n", sum.Index, sum.PartitionID, sum.Checksum, sum.KeyCount, sum.IDCount) + _, _ = hasher.Write([]byte(sum.Checksum)) + } + + var buf [16]byte + _, _ = hasher.Digest().Read(buf[0:]) + + fmt.Printf("all-checksum = blake3-%x\n", buf) + + if showFrags { + for _, idx := range indexes { + fmt.Printf("==============================\n") + fmt.Printf("index: %v\n", idx.Name()) + fmt.Printf("==============================\n") + idx.WriteFragmentChecksums(os.Stdout, showBits, showOpsLog) + } + } +} diff --git a/index.go b/index.go index 13eab7f85..25a31ac30 100644 --- a/index.go +++ b/index.go @@ -17,6 +17,7 @@ package pilosa import ( "context" "fmt" + "io" "io/ioutil" "os" "path/filepath" @@ -768,3 +769,103 @@ func (idx *Index) SliceOfShards(field, view, viewPath string) (sliceOfShards []u } return } + +type AllTranslatorSummary struct { + Sums []*TranslatorSummary +} + +func NewAllTranslatorSummary() *AllTranslatorSummary { + return &AllTranslatorSummary{} +} +func (ats *AllTranslatorSummary) Append(b *AllTranslatorSummary) { + ats.Sums = append(ats.Sums, b.Sums...) +} + +func (ats *AllTranslatorSummary) Sort() { + // return sorted by index then PartitionID then Field + sort.Slice(ats.Sums, func(i, j int) bool { + a := ats.Sums[i] + b := ats.Sums[j] + if a.Index < b.Index { + return true + } + if a.Index > b.Index { + return false + } + // INVAR: a.Index == b.Index + if a.PartitionID < b.PartitionID { + return true + } + if a.PartitionID > b.PartitionID { + return false + } + return a.Field < b.Field + }) +} + +// sums is only guaranteed to be sorted by (index, PartitionID, field) iff err returns nil +func (i *Index) ComputeTranslatorSummary(verbose bool) (ats *AllTranslatorSummary, err error) { + i.mu.RLock() + defer i.mu.RUnlock() + + ats = &AllTranslatorSummary{} + + fmt.Printf("\nindex: %v\n=================\n", i.name) + for _, fld := range i.fields { + sum, err := fld.translateStore.ComputeTranslatorSummary() + if err != nil { + return ats, err + } + sum.Field = fld.name + sum.Index = i.Name() + sum.Checksum = blake3sum16([]byte(fmt.Sprintf("%v/%v/%v", sum.Checksum, fld.name, i.Name()))) + + if verbose { + fmt.Printf("row blake3-%v keyN: %5v idN: %5v field: '%v'\n", sum.Checksum, sum.KeyCount, sum.IDCount, fld.name) + } + ats.Sums = append(ats.Sums, sum) + } + + fmt.Printf("====================\n") + + for partitionID, store := range i.translateStores { + sum, err := store.ComputeTranslatorSummary() + if err != nil { + return ats, err + } + if sum == nil { + // probably one of the Noop stores + continue + } + sum.PartitionID = partitionID + sum.Index = i.Name() + + sum.Checksum = blake3sum16([]byte(fmt.Sprintf("%v/%v/%v", sum.Checksum, partitionID, i.Name()))) + if verbose { + fmt.Printf("col blake3-%v keyN: %10v idN: %10v paritionID: %03v \n", sum.Checksum, sum.KeyCount, sum.IDCount, partitionID) + } + ats.Sums = append(ats.Sums, sum) + } + return +} + +func (idx *Index) WriteFragmentChecksums(w io.Writer, showBits, showOps bool) { + paths, err := listFilesUnderDir(idx.path, false, "", true) + panicOn(err) + index := idx.name + n := 0 + for _, relpath := range paths { + field, view, shard, err := fragmentSpecFromRoaringPath(relpath) + if err != nil { + continue // ignore .meta paths + } + abspath := idx.path + sep + relpath + + checksum, hotbits := RoaringFragmentChecksum(abspath, index, field, view, shard) + fmt.Fprintf(w, "frg blake3-%v field: '%v' view: '%v' shard: %3v hotbits: %10v\n", checksum, field, view, shard, hotbits) + n++ + } + if n == 0 { + fmt.Fprintf(w, "empty index '%v'", idx.path) + } +} diff --git a/mock/translator.go b/mock/translator.go index 8a28b504c..e577baf12 100644 --- a/mock/translator.go +++ b/mock/translator.go @@ -37,6 +37,10 @@ type TranslateStore struct { EntryReaderFunc func(ctx context.Context, offset uint64) (pilosa.TranslateEntryReader, error) } +func (s *TranslateStore) ComputeTranslatorSummary() (sum *pilosa.TranslatorSummary, err error) { + return +} + func (s *TranslateStore) Close() error { return s.CloseFunc() } diff --git a/rrtx.go b/rrtx.go index d5f841f46..49b1e7b28 100644 --- a/rrtx.go +++ b/rrtx.go @@ -39,7 +39,7 @@ func (tx *RoaringTx) Type() string { } func (tx *RoaringTx) Dump() { - fmt.Printf("%v\n", tx.Index.StringifiedRoaringKeys()) + fmt.Printf("%v\n", tx.Index.StringifiedRoaringKeys(false, false)) } func (tx *RoaringTx) UseRowCache() bool { diff --git a/translate.go b/translate.go index 4efa26129..101a854a3 100644 --- a/translate.go +++ b/translate.go @@ -89,6 +89,29 @@ type TranslateStore interface { // It should read from the reader and replace the data store with // the read payload. ReadFrom(io.Reader) (int64, error) + + ComputeTranslatorSummary() (sum *TranslatorSummary, err error) +} + +// TranslatorSummary is returned, for example from the boltdb string key translators, +// by calling ComputeTranslatorSummary(). Non-boltdb mocks, etc no-op that method. +type TranslatorSummary struct { + Index string + + // ParitionID is filled for column keys + PartitionID int + + // Field is filled for row keys + Field string + + // Checksum has a blake3 crypto hash of all the keys->ID and all the ID->key mappings + Checksum string + + // KeyCount has the number of Key->ID mappings + KeyCount int + + // IDCount has the number of ID->Key mappings + IDCount int } // OpenTranslateStoreFunc represents a function for instantiating and opening a TranslateStore. @@ -289,6 +312,9 @@ func OpenInMemTranslateStore(rawurl, index, field string, partitionID, partition return NewInMemTranslateStore(index, field, partitionID, partitionN), nil } +func (s *InMemTranslateStore) ComputeTranslatorSummary() (sum *TranslatorSummary, err error) { + panic("TODO") +} func (s *InMemTranslateStore) Close() error { return nil } diff --git a/txfactory.go b/txfactory.go index 0d4dcd7a9..e9231ab66 100644 --- a/txfactory.go +++ b/txfactory.go @@ -16,6 +16,7 @@ package pilosa import ( "fmt" + "io" "os" "path/filepath" "strconv" @@ -26,6 +27,7 @@ import ( "github.com/pilosa/pilosa/v2/roaring" "github.com/pilosa/pilosa/v2/txkey" "github.com/pkg/errors" + "github.com/zeebo/blake3" ) // public strings that pilosa/server/config.go can reference @@ -394,7 +396,9 @@ func fragmentSpecFromRoaringPath(path string) (field, view string, shard uint64, return } -func (idx *Index) StringifiedRoaringKeys() (r string) { +// hashOnly means only show the value hash, not the content bits. +// showOps means display the ops log. +func (idx *Index) StringifiedRoaringKeys(hashOnly, showOps bool) (r string) { paths, err := listFilesUnderDir(idx.path, false, "", true) panicOn(err) index := idx.name @@ -407,8 +411,8 @@ func (idx *Index) StringifiedRoaringKeys() (r string) { continue // ignore .meta paths } abspath := idx.path + sep + relpath - const showOps = false - s, err := stringifiedRawRoaringFragment(abspath, index, field, view, shard, showOps) + + s, _, err := stringifiedRawRoaringFragment(abspath, index, field, view, shard, showOps, hashOnly, os.Stdout) panicOn(err) //r += fmt.Sprintf("path:'%v' fragment contains:\n") + s if s == "" { @@ -426,7 +430,20 @@ func (idx *Index) StringifiedRoaringKeys() (r string) { return "roaring-" + r } -func stringifiedRawRoaringFragment(path string, index, field, view string, shard uint64, showOps bool) (r string, err error) { +func RoaringFragmentChecksum(path string, index, field, view string, shard uint64) (r string, hotbits int) { + hasher := blake3.New() + showOps := false + hashOnly := true + _, hotbits, err := stringifiedRawRoaringFragment(path, index, field, view, shard, showOps, hashOnly, hasher) + panicOn(err) + fmt.Fprintf(hasher, "%v/%v/%v/%v", index, field, view, shard) + var buf [16]byte + _, _ = hasher.Digest().Read(buf[0:]) + return fmt.Sprintf("%x", buf), hotbits + +} + +func stringifiedRawRoaringFragment(path string, index, field, view string, shard uint64, showOps, hashOnly bool, w io.Writer) (r string, hotbits int, err error) { var info roaring.BitmapInfo _ = info @@ -473,10 +490,10 @@ func stringifiedRawRoaringFragment(path string, index, field, view string, shard to: info.To, } if info.ContainerCount > 0 { - printContainers(info, pC) + printContainers(w, info, pC) } if info.Ops > 0 { - printOps(info) + printOps(w, info) } } @@ -491,13 +508,19 @@ func stringifiedRawRoaringFragment(path string, index, field, view string, shard cts := roaring.NewSliceContainers() cts.Put(ckey, ct) rbm := &roaring.Bitmap{Containers: cts} - srbm := bitmapAsString(rbm) - panicOn(err) + var srbm string + if !hashOnly { + srbm = bitmapAsString(rbm) + } bkey := string(txkey.Key(index, field, view, shard, ckey)) - r += fmt.Sprintf("%v -> %v (%v hot)\n", bkey, hash, ct.N()) - r += " ......." + srbm + "\n" + n := ct.N() + hotbits += int(n) + r += fmt.Sprintf("%v -> %v (%v hot)\n", bkey, hash, n) + if !hashOnly { + r += " ......." + srbm + "\n" + } } return @@ -579,9 +602,9 @@ type pointerContext struct { from, to uintptr } -func printOps(info roaring.BitmapInfo) { - fmt.Fprintln(os.Stdout, " Ops:") - tw := tabwriter.NewWriter(os.Stdout, 0, 8, 0, '\t', 0) +func printOps(w io.Writer, info roaring.BitmapInfo) { + fmt.Fprintln(w, " Ops:") + tw := tabwriter.NewWriter(w, 0, 8, 0, '\t', 0) fmt.Fprintf(tw, " \t%s\t%s\t%s\t\n", "TYPE", "OpN", "SIZE") printed := 0 for _, op := range info.OpDetails { @@ -606,9 +629,9 @@ func (p *pointerContext) pretty(c roaring.ContainerInfo) string { } // stolen from ctl/inspect.go -func printContainers(info roaring.BitmapInfo, pC pointerContext) { - fmt.Fprintln(os.Stdout, " Containers:") - tw := tabwriter.NewWriter(os.Stdout, 0, 8, 0, '\t', 0) +func printContainers(w io.Writer, info roaring.BitmapInfo, pC pointerContext) { + fmt.Fprintln(w, " Containers:") + tw := tabwriter.NewWriter(w, 0, 8, 0, '\t', 0) fmt.Fprintf(tw, " \t\tRoaring\t\t\t\tOps\t\t\t\tFlags\t\n") fmt.Fprintf(tw, "\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t\n", "KEY", "TYPE", "N", "ALLOC", "OFFSET", "TYPE", "N", "ALLOC", "OFFSET", "FLAGS") c1s := info.Containers