luxfi/zapdb
1
0
Fork 0
mirror of https://github.com/luxfi/zapdb.git synced 2026-07-27 06:54:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

[BREAKING] feat(index): Use flatbuffers instead of protobuf (#1546)

Browse Source 
This PR 
- Uses flatbuffers instead of protobufs for table index and directly stores byte slices in the cache.
- Uses MaxVersion to pick the oldest tables for compaction first.
- Uses leveldb/bloom so that we can test it without unmarshal
- Adds uncompressed size and key count in table index.
- Updates write bench tool to use managed mode.
This commit is contained in:
Ibrahim Jarif
2020-10-03 22:04:58 +05:30
committed by GitHub
parent 7c1e905a5d
commit ce11ae2977
28 changed files with 1209 additions and 979 deletions
Download Patch File Download Diff File Expand all files Collapse all files
badger/cmd/info.go
+25 -5
View File
@@ -22,6 +22,7 @@ import (
"fmt"
"io/ioutil"
"os"
"path"
"path/filepath"
"sort"
"strings"
@@ -239,27 +240,46 @@ func dur(src, dst time.Time) string {
return humanize.RelTime(dst, src, "earlier", "later")
}
func getInfo(fileInfos []os.FileInfo, tid uint64) int64 {
fileName := table.IDToFilename(tid)
for _, fi := range fileInfos {
if path.Base(fi.Name()) == fileName {
return fi.Size()
}
}
return 0
}
func tableInfo(dir, valueDir string, db *badger.DB) {
// we want all tables with keys count here.
tables := db.Tables(true)
tables := db.Tables()
fileInfos, err := ioutil.ReadDir(dir)
y.Check(err)
fmt.Println()
fmt.Println("SSTable [Li, Id, Total Keys including internal keys] " +
"[Left Key, Version -> Right Key, Version] [Index Size] [BF Size]")
"[Compression Ratio, Uncompressed Size, Index Size, BF Size] " +
"[Left Key, Version -> Right Key, Version]")
totalIndex := uint64(0)
totalBloomFilter := uint64(0)
totalCompressionRatio := float64(0.0)
for _, t := range tables {
lk, lt := y.ParseKey(t.Left), y.ParseTs(t.Left)
rk, rt := y.ParseKey(t.Right), y.ParseTs(t.Right)
compressionRatio := float64(t.UncompressedSize) /
float64(getInfo(fileInfos, t.ID)-int64(t.IndexSz))
fmt.Printf("SSTable [L%d, %03d, %07d] [%.2f, %s, %s, %s] [%20X, v%d -> %20X, v%d]\n",
t.Level, t.ID, t.KeyCount, compressionRatio, hbytes(int64(t.UncompressedSize)),
hbytes(int64(t.IndexSz)), hbytes(int64(t.BloomFilterSize)), lk, lt, rk, rt)
totalIndex += uint64(t.IndexSz)
totalBloomFilter += uint64(t.BloomFilterSize)
fmt.Printf("SSTable [L%d, %03d, %07d] [%20X, v%d -> %20X, v%d] [%s] [%s] \n",
t.Level, t.ID, t.KeyCount, lk, lt, rk, rt, hbytes(int64(t.IndexSz)),
hbytes(int64(t.BloomFilterSize)))
totalCompressionRatio += compressionRatio
}
fmt.Println()
fmt.Printf("Total Index Size: %s\n", hbytes(int64(totalIndex)))
fmt.Printf("Total BloomFilter Size: %s\n", hbytes(int64(totalIndex)))
fmt.Printf("Mean Compression Ratio: %.2f\n", totalCompressionRatio/float64(len(tables)))
fmt.Println()
}
badger/cmd/read_bench.go
+15 -36
View File
@@ -19,6 +19,7 @@ package cmd
import (
"context"
"fmt"
"math"
"math/rand"
"strings"
"sync/atomic"
@@ -120,7 +121,6 @@ func readBench(cmd *cobra.Command, args []string) error {
return y.Wrapf(err, "unable to open DB")
}
defer db.Close()
now := time.Now()
fmt.Println("*********************************************************")
fmt.Println("Starting to benchmark Reads")
@@ -131,32 +131,7 @@ func readBench(cmd *cobra.Command, args []string) error {
fullScanDB(db)
return nil
}
keys, err := getSampleKeys(db)
if err != nil {
return y.Wrapf(err, "error while sampling keys")
}
fmt.Println("*********************************************************")
fmt.Printf("Total Sampled Keys: %d, read in time: %s\n", len(keys), time.Since(now))
fmt.Println("*********************************************************")
if len(keys) == 0 {
fmt.Println("DB is empty, hence returning")
return nil
}
c := z.NewCloser(0)
startTime = time.Now()
for i := 0; i < numGoroutines; i++ {
c.AddRunning(1)
go readKeys(db, c, keys)
}
// also start printing stats
c.AddRunning(1)
go printStats(c)
<-time.After(dur)
c.SignalAndWait()
readTest(db, dur)
return nil
}
@@ -199,16 +174,20 @@ func readKeys(db *badger.DB, c *z.Closer, keys [][]byte) {
func lookupForKey(db *badger.DB, key []byte) (sz uint64) {
err := db.View(func(txn *badger.Txn) error {
itm, err := txn.Get(key)
y.Check(err)
iopt := badger.DefaultIteratorOptions
iopt.AllVersions = true
it := txn.NewKeyIterator(key, iopt)
defer it.Close()
if keysOnly {
sz = uint64(itm.KeySize())
} else {
y.Check2(itm.ValueCopy(nil))
sz = uint64(itm.EstimatedSize())
cnt := 0
for it.Seek(key); it.Valid(); it.Next() {
itm := it.Item()
sz += uint64(itm.EstimatedSize())
cnt++
if cnt == 10 {
break
}
}
return nil
})
y.Check(err)
@@ -219,7 +198,7 @@ func lookupForKey(db *badger.DB, key []byte) (sz uint64) {
func getSampleKeys(db *badger.DB) ([][]byte, error) {
var keys [][]byte
count := 0
stream := db.NewStream()
stream := db.NewStreamAt(math.MaxUint64)
// overide stream.KeyToList as we only want keys. Also
// we can take only first version for the key.
badger/cmd/write_bench.go
+73 -12
View File
@@ -21,6 +21,7 @@ import (
"encoding/binary"
"fmt"
"log"
"math"
"math/rand"
"os"
"path/filepath"
@@ -88,7 +89,7 @@ const (
func init() {
benchCmd.AddCommand(writeBenchCmd)
writeBenchCmd.Flags().IntVarP(&keySz, "key-size", "k", 32, "Size of key")
writeBenchCmd.Flags().IntVarP(&valSz, "val-size", "v", 128, "Size of value")
writeBenchCmd.Flags().IntVar(&valSz, "val-size", 128, "Size of value")
writeBenchCmd.Flags().Float64VarP(&numKeys, "keys-mil", "m", 10.0,
"Number of keys to add in millions")
writeBenchCmd.Flags().BoolVar(&syncWrites, "sync", true,
@@ -96,12 +97,12 @@ func init() {
writeBenchCmd.Flags().BoolVarP(&force, "force-compact", "f", true,
"Force compact level 0 on close.")
writeBenchCmd.Flags().BoolVarP(&sorted, "sorted", "s", false, "Write keys in sorted order.")
writeBenchCmd.Flags().BoolVarP(&showLogs, "logs", "l", false, "Show Badger logs.")
writeBenchCmd.Flags().BoolVarP(&showLogs, "verbose", "v", false, "Show Badger logs.")
writeBenchCmd.Flags().IntVarP(&valueThreshold, "value-th", "t", 1<<10, "Value threshold")
writeBenchCmd.Flags().IntVarP(&numVersions, "num-version", "n", 1, "Number of versions to keep")
writeBenchCmd.Flags().Int64Var(&blockCacheSize, "block-cache", 0,
writeBenchCmd.Flags().Int64Var(&blockCacheSize, "block-cache-mb", 0,
"Size of block cache in MB")
writeBenchCmd.Flags().Int64Var(&indexCacheSize, "index-cache", 0,
writeBenchCmd.Flags().Int64Var(&indexCacheSize, "index-cache-mb", 0,
"Size of index cache in MB.")
writeBenchCmd.Flags().Uint32Var(&vlogMaxEntries, "vlog-maxe", 1000000, "Value log Max Entries")
writeBenchCmd.Flags().StringVarP(&encryptionKey, "encryption-key", "e", "",
@@ -112,7 +113,7 @@ func init() {
"Load Bloom filter on DB open.")
writeBenchCmd.Flags().BoolVar(&detectConflicts, "conficts", true,
"If true, it badger will detect the conflicts")
writeBenchCmd.Flags().BoolVar(&compression, "compression", false,
writeBenchCmd.Flags().BoolVar(&compression, "compression", true,
"If true, badger will use ZSTD mode")
writeBenchCmd.Flags().BoolVar(&showDir, "show-dir", false,
"If true, the report will include the directory contents")
@@ -133,7 +134,7 @@ func writeRandom(db *badger.DB, num uint64) error {
y.Check2(rand.Read(value))
es := uint64(keySz + valSz) // entry size is keySz + valSz
batch := db.NewWriteBatch()
batch := db.NewManagedWriteBatch()
ttlPeriod, errParse := time.ParseDuration(ttlDuration)
y.Check(errParse)
@@ -146,11 +147,11 @@ func writeRandom(db *badger.DB, num uint64) error {
if ttlPeriod != 0 {
e.WithTTL(ttlPeriod)
}
err := batch.SetEntry(e)
err := batch.SetEntryAt(e, 1)
for err == badger.ErrBlockedWrites {
time.Sleep(time.Second)
batch = db.NewWriteBatch()
err = batch.SetEntry(e)
batch = db.NewManagedWriteBatch()
err = batch.SetEntryAt(e, 1)
}
if err != nil {
panic(err)
@@ -162,6 +163,34 @@ func writeRandom(db *badger.DB, num uint64) error {
return batch.Flush()
}
func readTest(db *badger.DB, dur time.Duration) {
now := time.Now()
keys, err := getSampleKeys(db)
if err != nil {
panic(err)
}
fmt.Println("*********************************************************")
fmt.Printf("Total Sampled Keys: %d, read in time: %s\n", len(keys), time.Since(now))
fmt.Println("*********************************************************")
if len(keys) == 0 {
fmt.Println("DB is empty, hence returning")
return
}
c := z.NewCloser(0)
readStartTime := time.Now()
for i := 0; i < numGoroutines; i++ {
c.AddRunning(1)
go readKeys(db, c, keys)
}
// also start printing stats
c.AddRunning(1)
go printReadStats(c, readStartTime)
<-time.After(dur)
c.SignalAndWait()
}
func writeSorted(db *badger.DB, num uint64) error {
value := make([]byte, valSz)
y.Check2(rand.Read(value))
@@ -257,7 +286,7 @@ func writeBench(cmd *cobra.Command, args []string) error {
}
fmt.Printf("Opening badger with options = %+v\n", opt)
db, err := badger.Open(opt)
db, err := badger.OpenManaged(opt)
if err != nil {
return err
}
@@ -282,6 +311,16 @@ func writeBench(cmd *cobra.Command, args []string) error {
if sorted {
err = writeSorted(db, num)
} else {
go func() {
for {
select {
case <-c.HasBeenClosed():
return
case <-time.After(30 * time.Second):
readTest(db, 5*time.Minute)
}
}
}()
err = writeRandom(db, num)
}
@@ -296,7 +335,7 @@ func showKeysStats(db *badger.DB) {
validKeyCount uint32
)
txn := db.NewTransaction(false)
txn := db.NewTransactionAt(math.MaxUint64, false)
defer txn.Discard()
iopt := badger.DefaultIteratorOptions
@@ -363,7 +402,7 @@ func reportStats(c *z.Closer, db *badger.DB) {
entries := atomic.LoadUint64(&entriesWritten)
bytesRate := sz / uint64(dur.Seconds())
entriesRate := entries / uint64(dur.Seconds())
fmt.Printf("Time elapsed: %s, bytes written: %s, speed: %s/sec, "+
fmt.Printf("[WRITE] Time elapsed: %s, bytes written: %s, speed: %s/sec, "+
"entries written: %d, speed: %d/sec, gcSuccess: %d\n", y.FixedDuration(time.Since(startTime)),
humanize.Bytes(sz), humanize.Bytes(bytesRate), entries, entriesRate, gcSuccess)
}
@@ -453,3 +492,25 @@ func dropPrefix(c *z.Closer, db *badger.DB) {
}
}
}
func printReadStats(c *z.Closer, startTime time.Time) {
defer c.Done()
t := time.NewTicker(time.Second)
defer t.Stop()
for {
select {
case <-c.HasBeenClosed():
return
case <-t.C:
dur := time.Since(startTime)
sz := atomic.LoadUint64(&sizeRead)
entries := atomic.LoadUint64(&entriesRead)
bytesRate := sz / uint64(dur.Seconds())
entriesRate := entries / uint64(dur.Seconds())
fmt.Printf("[READ] Time elapsed: %s, bytes read: %s, speed: %s/sec, "+
"entries read: %d, speed: %d/sec\n", y.FixedDuration(time.Since(startTime)),
humanize.Bytes(sz), humanize.Bytes(bytesRate), entries, entriesRate)
}
}
}
compaction.go
-15
View File
@@ -23,8 +23,6 @@ import (
"math"
"sync"
"golang.org/x/net/trace"
"github.com/dgraph-io/badger/v2/table"
"github.com/dgraph-io/badger/v2/y"
)
@@ -128,19 +126,6 @@ type compactStatus struct {
levels []*levelCompactStatus
}
func (cs *compactStatus) toLog(tr trace.Trace) {
cs.RLock()
defer cs.RUnlock()
tr.LazyPrintf("Compaction status:")
for i, l := range cs.levels {
if l.debug() == "" {
continue
}
tr.LazyPrintf("[%d] %s", i, l.debug())
}
}
func (cs *compactStatus) overlapsWith(level int, this keyRange) bool {
cs.RLock()
defer cs.RUnlock()
db.go
+52 -11
View File
@@ -54,12 +54,13 @@ const (
)
type closers struct {
updateSize *z.Closer
compactors *z.Closer
memtable *z.Closer
writes *z.Closer
valueGC *z.Closer
pub *z.Closer
updateSize *z.Closer
compactors *z.Closer
memtable *z.Closer
writes *z.Closer
valueGC *z.Closer
pub *z.Closer
cacheHealth *z.Closer
}
// DB provides the various functions required to interact with Badger.
@@ -249,8 +250,7 @@ func checkAndSetOptions(opt *Options) error {
needCache := (opt.Compression != options.None) || (len(opt.EncryptionKey) > 0)
if needCache && opt.BlockCacheSize == 0 {
opt.Warningf("BlockCacheSize should be set " +
"since compression/encryption are enabled")
panic("BlockCacheSize should be set since compression/encryption are enabled")
}
return nil
}
@@ -362,6 +362,10 @@ func Open(opt Options) (db *DB, err error) {
return nil, errors.Wrap(err, "failed to create bf cache")
}
}
db.closers.cacheHealth = z.NewCloser(1)
go db.monitorCache(db.closers.cacheHealth)
if db.opt.InMemory {
db.opt.SyncWrites = false
// If badger is running in memory mode, push everything into the LSM Tree.
@@ -437,6 +441,42 @@ func Open(opt Options) (db *DB, err error) {
return db, nil
}
func (db *DB) monitorCache(c *z.Closer) {
defer c.Done()
if db.blockCache == nil {
return
}
count := 0
analyze := func(name string, metrics *ristretto.Metrics) {
// If the mean life expectancy is less than 10 seconds, the cache
// might be too small.
le := metrics.LifeExpectancySeconds()
lifeTooShort := le.Count > 0 && float64(le.Sum)/float64(le.Count) < 10
hitRatioTooLow := metrics.Ratio() > 0 && metrics.Ratio() < 0.4
if lifeTooShort && hitRatioTooLow {
db.opt.Warningf("%s might be too small. Metrics: %s\n", name, metrics)
db.opt.Warningf("Cache life expectancy (in seconds): %+v\n", le)
} else if le.Count > 1000 && count%5 == 0 {
db.opt.Infof("%s metrics: %s\n", name, metrics)
}
}
ticker := time.NewTicker(1 * time.Minute)
defer ticker.Stop()
for {
select {
case <-c.HasBeenClosed():
return
case <-ticker.C:
}
count++
analyze("Block cache", db.BlockCacheMetrics())
analyze("Index cache", db.IndexCacheMetrics())
}
}
// getHead prints all the head pointer in the DB and return the max value.
func (db *DB) getHead() (valuePointer, uint64) {
// This is a hack. If we use newTransaction(..) we'll end up in deadlock
@@ -562,6 +602,7 @@ func (db *DB) close() (err error) {
close(db.writeCh)
db.closers.pub.SignalAndWait()
db.closers.cacheHealth.Signal()
// Now close the value log.
if vlogErr := db.vlog.Close(); vlogErr != nil {
@@ -1450,15 +1491,15 @@ func (db *DB) GetSequence(key []byte, bandwidth uint64) (*Sequence, error) {
// Tables gets the TableInfo objects from the level controller. If withKeysCount
// is true, TableInfo objects also contain counts of keys for the tables.
func (db *DB) Tables(withKeysCount bool) []TableInfo {
return db.lc.getTableInfo(withKeysCount)
func (db *DB) Tables() []TableInfo {
return db.lc.getTableInfo()
}
// KeySplits can be used to get rough key ranges to divide up iteration over
// the DB.
func (db *DB) KeySplits(prefix []byte) []string {
var splits []string
tables := db.Tables(false)
tables := db.Tables()
// We just want table ranges here and not keys count.
for _, ti := range tables {
db2_test.go
+13 -10
View File
@@ -495,26 +495,29 @@ func TestCompactionFilePicking(t *testing.T) {
}
tables := db.lc.levels[2].tables
db.lc.sortByOverlap(tables, cdef)
db.lc.sortByHeuristic(tables, cdef)
var expKey [8]byte
// First table should be with smallest and biggest keys as 1 and 4.
// First table should be with smallest and biggest keys as 1 and 4 which
// has the lowest version.
binary.BigEndian.PutUint64(expKey[:], uint64(1))
require.Equal(t, expKey[:], y.ParseKey(tables[0].Smallest()))
binary.BigEndian.PutUint64(expKey[:], uint64(4))
require.Equal(t, expKey[:], y.ParseKey(tables[0].Biggest()))
// Second table should be with smallest and biggest keys as 13 and 18.
// Second table should be with smallest and biggest keys as 13 and 18
// which has the second lowest version.
binary.BigEndian.PutUint64(expKey[:], uint64(13))
require.Equal(t, expKey[:], y.ParseKey(tables[1].Smallest()))
binary.BigEndian.PutUint64(expKey[:], uint64(18))
require.Equal(t, expKey[:], y.ParseKey(tables[1].Biggest()))
// Third table should be with smallest and biggest keys as 5 and 12.
binary.BigEndian.PutUint64(expKey[:], uint64(5))
require.Equal(t, expKey[:], y.ParseKey(tables[2].Smallest()))
binary.BigEndian.PutUint64(expKey[:], uint64(12))
binary.BigEndian.PutUint64(expKey[:], uint64(18))
require.Equal(t, expKey[:], y.ParseKey(tables[2].Biggest()))
// Third table should be with smallest and biggest keys as 5 and 12 which
// has the maximum version.
binary.BigEndian.PutUint64(expKey[:], uint64(5))
require.Equal(t, expKey[:], y.ParseKey(tables[1].Smallest()))
binary.BigEndian.PutUint64(expKey[:], uint64(12))
require.Equal(t, expKey[:], y.ParseKey(tables[1].Biggest()))
}
// addToManifest function is used in TestCompactionFilePicking. It adds table to db manifest.
db_test.go
+4 -1
View File
@@ -380,7 +380,7 @@ func TestStreamDB(t *testing.T) {
dir, err := ioutil.TempDir("", "badger-test")
require.NoError(t, err)
defer removeDir(dir)
opts := getTestOptions(dir).WithCompression(options.ZSTD)
opts := getTestOptions(dir).WithCompression(options.ZSTD).WithBlockCacheSize(100 << 20)
db, err := OpenManaged(opts)
require.NoError(t, err)
@@ -838,6 +838,7 @@ func TestLoad(t *testing.T) {
require.NoError(t, err)
opt := getTestOptions("")
opt.EncryptionKey = key
opt.BlockCacheSize = 100 << 20
opt.Compression = options.None
testLoad(t, opt)
})
@@ -848,11 +849,13 @@ func TestLoad(t *testing.T) {
opt := getTestOptions("")
opt.EncryptionKey = key
opt.Compression = options.ZSTD
opt.BlockCacheSize = 100 << 20
testLoad(t, opt)
})
t.Run("TestLoad without Encryption and with compression", func(t *testing.T) {
opt := getTestOptions("")
opt.Compression = options.ZSTD
opt.BlockCacheSize = 100 << 20
testLoad(t, opt)
})
}
fb/BlockOffset.go
+104
View File
@@ -0,0 +1,104 @@
// Code generated by the FlatBuffers compiler. DO NOT EDIT.
package fb
import (
flatbuffers "github.com/google/flatbuffers/go"
)
type BlockOffset struct {
_tab flatbuffers.Table
}
func GetRootAsBlockOffset(buf []byte, offset flatbuffers.UOffsetT) *BlockOffset {
n := flatbuffers.GetUOffsetT(buf[offset:])
x := &BlockOffset{}
x.Init(buf, n+offset)
return x
}
func (rcv *BlockOffset) Init(buf []byte, i flatbuffers.UOffsetT) {
rcv._tab.Bytes = buf
rcv._tab.Pos = i
}
func (rcv *BlockOffset) Table() flatbuffers.Table {
return rcv._tab
}
func (rcv *BlockOffset) Key(j int) byte {
o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
if o != 0 {
a := rcv._tab.Vector(o)
return rcv._tab.GetByte(a + flatbuffers.UOffsetT(j*1))
}
return 0
}
func (rcv *BlockOffset) KeyLength() int {
o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
if o != 0 {
return rcv._tab.VectorLen(o)
}
return 0
}
func (rcv *BlockOffset) KeyBytes() []byte {
o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
if o != 0 {
return rcv._tab.ByteVector(o + rcv._tab.Pos)
}
return nil
}
func (rcv *BlockOffset) MutateKey(j int, n byte) bool {
o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
if o != 0 {
a := rcv._tab.Vector(o)
return rcv._tab.MutateByte(a+flatbuffers.UOffsetT(j*1), n)
}
return false
}
func (rcv *BlockOffset) Offset() uint32 {
o := flatbuffers.UOffsetT(rcv._tab.Offset(6))
if o != 0 {
return rcv._tab.GetUint32(o + rcv._tab.Pos)
}
return 0
}
func (rcv *BlockOffset) MutateOffset(n uint32) bool {
return rcv._tab.MutateUint32Slot(6, n)
}
func (rcv *BlockOffset) Len() uint32 {
o := flatbuffers.UOffsetT(rcv._tab.Offset(8))
if o != 0 {
return rcv._tab.GetUint32(o + rcv._tab.Pos)
}
return 0
}
func (rcv *BlockOffset) MutateLen(n uint32) bool {
return rcv._tab.MutateUint32Slot(8, n)
}
func BlockOffsetStart(builder *flatbuffers.Builder) {
builder.StartObject(3)
}
func BlockOffsetAddKey(builder *flatbuffers.Builder, key flatbuffers.UOffsetT) {
builder.PrependUOffsetTSlot(0, flatbuffers.UOffsetT(key), 0)
}
func BlockOffsetStartKeyVector(builder *flatbuffers.Builder, numElems int) flatbuffers.UOffsetT {
return builder.StartVector(1, numElems, 1)
}
func BlockOffsetAddOffset(builder *flatbuffers.Builder, offset uint32) {
builder.PrependUint32Slot(1, offset, 0)
}
func BlockOffsetAddLen(builder *flatbuffers.Builder, len uint32) {
builder.PrependUint32Slot(2, len, 0)
}
func BlockOffsetEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT {
return builder.EndObject()
}
fb/TableIndex.go
+160
View File
@@ -0,0 +1,160 @@
// Code generated by the FlatBuffers compiler. DO NOT EDIT.
package fb
import (
flatbuffers "github.com/google/flatbuffers/go"
)
type TableIndex struct {
_tab flatbuffers.Table
}
func GetRootAsTableIndex(buf []byte, offset flatbuffers.UOffsetT) *TableIndex {
n := flatbuffers.GetUOffsetT(buf[offset:])
x := &TableIndex{}
x.Init(buf, n+offset)
return x
}
func (rcv *TableIndex) Init(buf []byte, i flatbuffers.UOffsetT) {
rcv._tab.Bytes = buf
rcv._tab.Pos = i
}
func (rcv *TableIndex) Table() flatbuffers.Table {
return rcv._tab
}
func (rcv *TableIndex) Offsets(obj *BlockOffset, j int) bool {
o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
if o != 0 {
x := rcv._tab.Vector(o)
x += flatbuffers.UOffsetT(j) * 4
x = rcv._tab.Indirect(x)
obj.Init(rcv._tab.Bytes, x)
return true
}
return false
}
func (rcv *TableIndex) OffsetsLength() int {
o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
if o != 0 {
return rcv._tab.VectorLen(o)
}
return 0
}
func (rcv *TableIndex) BloomFilter(j int) byte {
o := flatbuffers.UOffsetT(rcv._tab.Offset(6))
if o != 0 {
a := rcv._tab.Vector(o)
return rcv._tab.GetByte(a + flatbuffers.UOffsetT(j*1))
}
return 0
}
func (rcv *TableIndex) BloomFilterLength() int {
o := flatbuffers.UOffsetT(rcv._tab.Offset(6))
if o != 0 {
return rcv._tab.VectorLen(o)
}
return 0
}
func (rcv *TableIndex) BloomFilterBytes() []byte {
o := flatbuffers.UOffsetT(rcv._tab.Offset(6))
if o != 0 {
return rcv._tab.ByteVector(o + rcv._tab.Pos)
}
return nil
}
func (rcv *TableIndex) MutateBloomFilter(j int, n byte) bool {
o := flatbuffers.UOffsetT(rcv._tab.Offset(6))
if o != 0 {
a := rcv._tab.Vector(o)
return rcv._tab.MutateByte(a+flatbuffers.UOffsetT(j*1), n)
}
return false
}
func (rcv *TableIndex) EstimatedSize() uint32 {
o := flatbuffers.UOffsetT(rcv._tab.Offset(8))
if o != 0 {
return rcv._tab.GetUint32(o + rcv._tab.Pos)
}
return 0
}
func (rcv *TableIndex) MutateEstimatedSize(n uint32) bool {
return rcv._tab.MutateUint32Slot(8, n)
}
func (rcv *TableIndex) MaxVersion() uint64 {
o := flatbuffers.UOffsetT(rcv._tab.Offset(10))
if o != 0 {
return rcv._tab.GetUint64(o + rcv._tab.Pos)
}
return 0
}
func (rcv *TableIndex) MutateMaxVersion(n uint64) bool {
return rcv._tab.MutateUint64Slot(10, n)
}
func (rcv *TableIndex) UncompressedSize() uint32 {
o := flatbuffers.UOffsetT(rcv._tab.Offset(12))
if o != 0 {
return rcv._tab.GetUint32(o + rcv._tab.Pos)
}
return 0
}
func (rcv *TableIndex) MutateUncompressedSize(n uint32) bool {
return rcv._tab.MutateUint32Slot(12, n)
}
func (rcv *TableIndex) KeyCount() uint32 {
o := flatbuffers.UOffsetT(rcv._tab.Offset(14))
if o != 0 {
return rcv._tab.GetUint32(o + rcv._tab.Pos)
}
return 0
}
func (rcv *TableIndex) MutateKeyCount(n uint32) bool {
return rcv._tab.MutateUint32Slot(14, n)
}
func TableIndexStart(builder *flatbuffers.Builder) {
builder.StartObject(6)
}
func TableIndexAddOffsets(builder *flatbuffers.Builder, offsets flatbuffers.UOffsetT) {
builder.PrependUOffsetTSlot(0, flatbuffers.UOffsetT(offsets), 0)
}
func TableIndexStartOffsetsVector(builder *flatbuffers.Builder, numElems int) flatbuffers.UOffsetT {
return builder.StartVector(4, numElems, 4)
}
func TableIndexAddBloomFilter(builder *flatbuffers.Builder, bloomFilter flatbuffers.UOffsetT) {
builder.PrependUOffsetTSlot(1, flatbuffers.UOffsetT(bloomFilter), 0)
}
func TableIndexStartBloomFilterVector(builder *flatbuffers.Builder, numElems int) flatbuffers.UOffsetT {
return builder.StartVector(1, numElems, 1)
}
func TableIndexAddEstimatedSize(builder *flatbuffers.Builder, estimatedSize uint32) {
builder.PrependUint32Slot(2, estimatedSize, 0)
}
func TableIndexAddMaxVersion(builder *flatbuffers.Builder, maxVersion uint64) {
builder.PrependUint64Slot(3, maxVersion, 0)
}
func TableIndexAddUncompressedSize(builder *flatbuffers.Builder, uncompressedSize uint32) {
builder.PrependUint32Slot(4, uncompressedSize, 0)
}
func TableIndexAddKeyCount(builder *flatbuffers.Builder, keyCount uint32) {
builder.PrependUint32Slot(5, keyCount, 0)
}
func TableIndexEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT {
return builder.EndObject()
}
fb/flatbuffer.fbs
+35
View File
@@ -0,0 +1,35 @@
/*
* Copyright 2020 Dgraph Labs, Inc. and Contributors
*
* 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.
*/
namespace fb;
table TableIndex {
offsets:[BlockOffset];
bloom_filter:[ubyte];
estimated_size:uint32;
max_version:uint64;
uncompressed_size:uint32;
key_count:uint32;
}
table BlockOffset {
key:[ubyte];
offset:uint;
len:uint;
}
root_type TableIndex;
root_type BlockOffset;
fb/gen.sh
Executable
+12
View File
@@ -0,0 +1,12 @@
#!/usr/bin/env bash
set -e
## Install flatc if not present
## ref. https://google.github.io/flatbuffers/flatbuffers_guide_building.html
command -v flatc > /dev/null || { ./install_flatbuffers.sh ; }
flatc --go flatbuffer.fbs
# Move files to the correct directory.
mv fb/* ./
rmdir fb
fb/install_flatbuffers.sh
Executable
+47
View File
@@ -0,0 +1,47 @@
#!/usr/bin/env bash
set -e
install_mac() {
command -v brew > /dev/null || \
{ echo "[ERROR]: 'brew' command not not found. Exiting" 1>&2; exit 1; }
brew install flatbuffers
}
install_linux() {
for CMD in curl cmake g++ make; do
command -v $CMD > /dev/null || \
{ echo "[ERROR]: '$CMD' command not not found. Exiting" 1>&2; exit 1; }
done
## Create Temp Build Directory
BUILD_DIR=$(mktemp -d)
pushd $BUILD_DIR
## Fetch Latest Tarball
LATEST_VERSION=$(curl -s https://api.github.com/repos/google/flatbuffers/releases/latest | grep -oP '(?<=tag_name": ")[^"]+')
curl -sLO https://github.com/google/flatbuffers/archive/$LATEST_VERSION.tar.gz
tar xf $LATEST_VERSION.tar.gz
## Build Binaries
cd flatbuffers-${LATEST_VERSION#v}
cmake -G "Unix Makefiles" -DCMAKE_BUILD_TYPE=Release
make
./flattests
cp flatc /usr/local/bin/flatc
## Cleanup Temp Build Directory
popd
rm -rf $BUILD_DIR
}
SYSTEM=$(uname -s)
case ${SYSTEM,,} in
linux)
sudo bash -c "$(declare -f install_linux); install_linux"
;;
darwin)
install_mac
;;
esac
go.mod
+4 -2
View File
@@ -2,14 +2,16 @@ module github.com/dgraph-io/badger/v2
go 1.12
// replace github.com/dgraph-io/ristretto => /home/mrjn/go/src/github.com/dgraph-io/ristretto
require (
github.com/DataDog/zstd v1.4.1
github.com/cespare/xxhash v1.1.0
github.com/dgraph-io/ristretto v0.0.4-0.20200930150433-e1609c8d4ca6
github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2
github.com/dgraph-io/ristretto v0.0.4-0.20201003140040-36af8a316fc8
github.com/dustin/go-humanize v1.0.0
github.com/golang/protobuf v1.3.1
github.com/golang/snappy v0.0.1
github.com/google/flatbuffers v1.12.0
github.com/kr/pretty v0.1.0 // indirect
github.com/pkg/errors v0.9.1
github.com/spaolacci/murmur3 v1.1.0 // indirect
go.sum
+4 -2
View File
@@ -13,8 +13,8 @@ github.com/cpuguy83/go-md2man v1.0.10/go.mod h1:SmD6nW6nTyfqj6ABTjUi3V3JVMnlJmwc
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/dgraph-io/ristretto v0.0.4-0.20200930150433-e1609c8d4ca6 h1:maqkH6zkcUx1PP+LpbLMef2IOSz2/mLYT3RqecFcdDI=
github.com/dgraph-io/ristretto v0.0.4-0.20200930150433-e1609c8d4ca6/go.mod h1:bDI4cDaalvYSji3vBVDKrn9ouDZrwN974u8ZO/AhYXs=
github.com/dgraph-io/ristretto v0.0.4-0.20201003140040-36af8a316fc8 h1:LBKxHZx6GX+vkwoXO7Zd69/VSvyB3BMDBilnqQ4c/Ws=
github.com/dgraph-io/ristretto v0.0.4-0.20201003140040-36af8a316fc8/go.mod h1:bDI4cDaalvYSji3vBVDKrn9ouDZrwN974u8ZO/AhYXs=
github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2 h1:tdlZCpZ/P9DhczCTSixgIKmwPv6+wP5DGjqLYw5SUiA=
github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw=
github.com/dustin/go-humanize v1.0.0 h1:VSnTsYCnlFHaM2/igO1h6X3HA71jcobQuxemgkq4zYo=
@@ -24,6 +24,8 @@ github.com/golang/protobuf v1.3.1 h1:YF8+flBXS5eO826T4nzqPrxfhQThhXl0YzfuUPu4SBg
github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
github.com/google/flatbuffers v1.12.0 h1:/PtAHvnBY4Kqnx/xCQ3OIV9uYcSFGScBsWI3Oogeh6w=
github.com/google/flatbuffers v1.12.0/go.mod h1:1AeVuKshWv4vARoZatz6mlQ0JxURH0Kv5+zNeJKJCa8=
github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
iterator.go
+4 -13
View File
@@ -27,7 +27,6 @@ import (
"github.com/dgraph-io/badger/v2/options"
"github.com/dgraph-io/badger/v2/table"
"github.com/dgryski/go-farm"
"github.com/dgraph-io/badger/v2/y"
)
@@ -332,7 +331,8 @@ func (opt *IteratorOptions) pickTable(t table.TableInterface) bool {
}
// Bloom filter lookup would only work if opt.Prefix does NOT have the read
// timestamp as part of the key.
if opt.prefixIsKey && t.DoesNotHave(farm.Fingerprint64(opt.Prefix)) {
if opt.prefixIsKey && t.DoesNotHave(y.Hash(opt.Prefix)) {
y.NumLSMBloomHits.Add("pickTable", 1)
return false
}
return true
@@ -365,19 +365,10 @@ func (opt *IteratorOptions) pickTables(all []*table.Table) []*table.Table {
}
var out []*table.Table
hash := farm.Fingerprint64(opt.Prefix)
for _, t := range filtered {
// When we encounter the first table whose smallest key is higher than
// opt.Prefix, we can stop.
if opt.compareToPrefix(t.Smallest()) > 0 {
return out
if opt.pickTable(t) {
out = append(out, t)
}
// opt.Prefix is actually the key. So, we can run bloom filter checks
// as well.
if t.DoesNotHave(hash) {
continue
}
out = append(out, t)
}
return out
}
iterator_test.go
+1 -1
View File
@@ -38,7 +38,7 @@ type tableMock struct {
func (tm *tableMock) Smallest() []byte { return tm.left }
func (tm *tableMock) Biggest() []byte { return tm.right }
func (tm *tableMock) DoesNotHave(hash uint64) bool { return false }
func (tm *tableMock) DoesNotHave(hash uint32) bool { return false }
func TestPickTables(t *testing.T) {
opt := DefaultIteratorOptions
level_handler.go
+1 -3
View File
@@ -21,8 +21,6 @@ import (
"sort"
"sync"
"github.com/dgryski/go-farm"
"github.com/dgraph-io/badger/v2/table"
"github.com/dgraph-io/badger/v2/y"
"github.com/pkg/errors"
@@ -259,7 +257,7 @@ func (s *levelHandler) get(key []byte) (y.ValueStruct, error) {
tables, decr := s.getTableForKey(key)
keyNoTs := y.ParseKey(key)
hash := farm.Fingerprint64(keyNoTs)
hash := y.Hash(keyNoTs)
var maxVs y.ValueStruct
for _, th := range tables {
if th.DoesNotHave(hash) {
levels.go
+36 -45
View File
@@ -18,6 +18,7 @@ package badger
import (
"bytes"
"encoding/hex"
"fmt"
"math/rand"
"os"
@@ -841,23 +842,16 @@ func (s *levelsController) fillTablesL0(cd *compactDef) bool {
return true
}
// sortByOverlap sorts tables in increasing order of overlap with next level.
func (s *levelsController) sortByOverlap(tables []*table.Table, cd *compactDef) {
// sortByHeuristic sorts tables in increasing order of MaxVersion, so we
// compact older tables first.
func (s *levelsController) sortByHeuristic(tables []*table.Table, cd *compactDef) {
if len(tables) == 0 || cd.nextLevel == nil {
return
}
tableOverlap := make([]int, len(tables))
for i := range tables {
// get key range for table
tableRange := getKeyRange(tables[i])
// get overlap with next level
left, right := cd.nextLevel.overlappingTables(levelHandlerRLocked{}, tableRange)
tableOverlap[i] = right - left
}
// Sort tables by max version. This is what RocksDB does.
sort.Slice(tables, func(i, j int) bool {
return tableOverlap[i] < tableOverlap[j]
return tables[i].MaxVersion() < tables[j].MaxVersion()
})
}
@@ -870,15 +864,14 @@ func (s *levelsController) fillTables(cd *compactDef) bool {
if len(tables) == 0 {
return false
}
// We want to pick files from current level in order of increasing overlap with next level
// tables. Idea here is to first compact file from current level which has least overlap with
// next level. This provides us better write amplification.
s.sortByOverlap(tables, cd)
// We pick tables, so we compact older tables first. This is similar to
// kOldestLargestSeqFirst in RocksDB.
s.sortByHeuristic(tables, cd)
for _, t := range tables {
cd.thisSize = t.Size()
cd.thisRange = getKeyRange(t)
// If we're already compacting this range, don't do anything.
if s.cstatus.overlapsWith(cd.thisLevel.level, cd.thisRange) {
continue
}
@@ -962,6 +955,13 @@ func (s *levelsController) runCompactDef(l int, cd compactDef) (err error) {
s.kv.opt.Infof("LOG Compact %d->%d, del %d tables, add %d tables, took %v\n",
thisLevel.level, nextLevel.level, len(cd.top)+len(cd.bot),
len(newTables), time.Since(timeStart))
if cd.thisLevel.level != 0 && len(newTables) > 2*s.kv.opt.LevelSizeMultiplier {
s.kv.opt.Infof("This Range (numTables: %d)\nLeft:\n%s\nRight:\n%s\n",
len(cd.top), hex.Dump(cd.thisRange.left), hex.Dump(cd.thisRange.right))
s.kv.opt.Infof("Next Range (numTables: %d)\nLeft:\n%s\nRight:\n%s\n",
len(cd.bot), hex.Dump(cd.nextRange.left), hex.Dump(cd.nextRange.right))
}
return nil
}
@@ -999,14 +999,12 @@ func (s *levelsController) doCompact(id int, p compactionPriority) error {
s.kv.opt.Infof("[Compactor: %d] Running compaction: %+v for level: %d\n",
id, p, cd.thisLevel.level)
s.cstatus.toLog(cd.elog)
if err := s.runCompactDef(l, cd); err != nil {
// This compaction couldn't be done successfully.
s.kv.opt.Warningf("[Compactor: %d] LOG Compact FAILED with error: %+v: %+v", id, err, cd)
return err
}
s.cstatus.toLog(cd.elog)
s.kv.opt.Infof("[Compactor: %d] Compaction for level: %d DONE", id, cd.thisLevel.level)
return nil
}
@@ -1128,38 +1126,31 @@ func (s *levelsController) appendIterators(
// TableInfo represents the information about a table.
type TableInfo struct {
ID uint64
Level int
Left []byte
Right []byte
KeyCount uint64 // Number of keys in the table
EstimatedSz uint64
IndexSz int
BloomFilterSize int
ID uint64
Level int
Left []byte
Right []byte
KeyCount uint32 // Number of keys in the table
EstimatedSz uint32
UncompressedSize uint32
IndexSz int
BloomFilterSize int
}
func (s *levelsController) getTableInfo(withKeysCount bool) (result []TableInfo) {
func (s *levelsController) getTableInfo() (result []TableInfo) {
for _, l := range s.levels {
l.RLock()
for _, t := range l.tables {
var count uint64
if withKeysCount {
it := t.NewIterator(table.NOCACHE)
for it.Rewind(); it.Valid(); it.Next() {
count++
}
it.Close()
}
info := TableInfo{
ID: t.ID(),
Level: l.level,
Left: t.Smallest(),
Right: t.Biggest(),
KeyCount: count,
EstimatedSz: t.EstimatedSize(),
IndexSz: t.IndexSize(),
BloomFilterSize: t.BloomFilterSize(),
ID: t.ID(),
Level: l.level,
Left: t.Smallest(),
Right: t.Biggest(),
KeyCount: t.KeyCount(),
EstimatedSz: t.EstimatedSize(),
IndexSz: t.IndexSize(),
BloomFilterSize: t.BloomFilterSize(),
UncompressedSize: t.UncompressedSize(),
}
result = append(result, info)
}
pb/badgerpb2.pb.go
+41 -587
View File
@@ -91,7 +91,7 @@ func (x Checksum_Algorithm) String() string {
}
func (Checksum_Algorithm) EnumDescriptor() ([]byte, []int) {
return fileDescriptor_e63e84f9f0d3998c, []int{6, 0}
return fileDescriptor_e63e84f9f0d3998c, []int{4, 0}
}
type KV struct {
@@ -381,132 +381,6 @@ func (m *ManifestChange) GetCompression() uint32 {
return 0
}
type BlockOffset struct {
Key []byte `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`
Offset uint32 `protobuf:"varint,2,opt,name=offset,proto3" json:"offset,omitempty"`
Len uint32 `protobuf:"varint,3,opt,name=len,proto3" json:"len,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
func (m *BlockOffset) Reset() { *m = BlockOffset{} }
func (m *BlockOffset) String() string { return proto.CompactTextString(m) }
func (*BlockOffset) ProtoMessage() {}
func (*BlockOffset) Descriptor() ([]byte, []int) {
return fileDescriptor_e63e84f9f0d3998c, []int{4}
}
func (m *BlockOffset) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
}
func (m *BlockOffset) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
if deterministic {
return xxx_messageInfo_BlockOffset.Marshal(b, m, deterministic)
} else {
b = b[:cap(b)]
n, err := m.MarshalToSizedBuffer(b)
if err != nil {
return nil, err
}
return b[:n], nil
}
}
func (m *BlockOffset) XXX_Merge(src proto.Message) {
xxx_messageInfo_BlockOffset.Merge(m, src)
}
func (m *BlockOffset) XXX_Size() int {
return m.Size()
}
func (m *BlockOffset) XXX_DiscardUnknown() {
xxx_messageInfo_BlockOffset.DiscardUnknown(m)
}
var xxx_messageInfo_BlockOffset proto.InternalMessageInfo
func (m *BlockOffset) GetKey() []byte {
if m != nil {
return m.Key
}
return nil
}
func (m *BlockOffset) GetOffset() uint32 {
if m != nil {
return m.Offset
}
return 0
}
func (m *BlockOffset) GetLen() uint32 {
if m != nil {
return m.Len
}
return 0
}
type TableIndex struct {
Offsets []*BlockOffset `protobuf:"bytes,1,rep,name=offsets,proto3" json:"offsets,omitempty"`
BloomFilter []byte `protobuf:"bytes,2,opt,name=bloom_filter,json=bloomFilter,proto3" json:"bloom_filter,omitempty"`
EstimatedSize uint64 `protobuf:"varint,3,opt,name=estimated_size,json=estimatedSize,proto3" json:"estimated_size,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
func (m *TableIndex) Reset() { *m = TableIndex{} }
func (m *TableIndex) String() string { return proto.CompactTextString(m) }
func (*TableIndex) ProtoMessage() {}
func (*TableIndex) Descriptor() ([]byte, []int) {
return fileDescriptor_e63e84f9f0d3998c, []int{5}
}
func (m *TableIndex) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
}
func (m *TableIndex) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
if deterministic {
return xxx_messageInfo_TableIndex.Marshal(b, m, deterministic)
} else {
b = b[:cap(b)]
n, err := m.MarshalToSizedBuffer(b)
if err != nil {
return nil, err
}
return b[:n], nil
}
}
func (m *TableIndex) XXX_Merge(src proto.Message) {
xxx_messageInfo_TableIndex.Merge(m, src)
}
func (m *TableIndex) XXX_Size() int {
return m.Size()
}
func (m *TableIndex) XXX_DiscardUnknown() {
xxx_messageInfo_TableIndex.DiscardUnknown(m)
}
var xxx_messageInfo_TableIndex proto.InternalMessageInfo
func (m *TableIndex) GetOffsets() []*BlockOffset {
if m != nil {
return m.Offsets
}
return nil
}
func (m *TableIndex) GetBloomFilter() []byte {
if m != nil {
return m.BloomFilter
}
return nil
}
func (m *TableIndex) GetEstimatedSize() uint64 {
if m != nil {
return m.EstimatedSize
}
return 0
}
type Checksum struct {
Algo Checksum_Algorithm `protobuf:"varint,1,opt,name=algo,proto3,enum=badgerpb2.Checksum_Algorithm" json:"algo,omitempty"`
Sum uint64 `protobuf:"varint,2,opt,name=sum,proto3" json:"sum,omitempty"`
@@ -519,7 +393,7 @@ func (m *Checksum) Reset() { *m = Checksum{} }
func (m *Checksum) String() string { return proto.CompactTextString(m) }
func (*Checksum) ProtoMessage() {}
func (*Checksum) Descriptor() ([]byte, []int) {
return fileDescriptor_e63e84f9f0d3998c, []int{6}
return fileDescriptor_e63e84f9f0d3998c, []int{4}
}
func (m *Checksum) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@@ -576,7 +450,7 @@ func (m *DataKey) Reset() { *m = DataKey{} }
func (m *DataKey) String() string { return proto.CompactTextString(m) }
func (*DataKey) ProtoMessage() {}
func (*DataKey) Descriptor() ([]byte, []int) {
return fileDescriptor_e63e84f9f0d3998c, []int{7}
return fileDescriptor_e63e84f9f0d3998c, []int{5}
}
func (m *DataKey) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@@ -641,8 +515,6 @@ func init() {
proto.RegisterType((*KVList)(nil), "badgerpb2.KVList")
proto.RegisterType((*ManifestChangeSet)(nil), "badgerpb2.ManifestChangeSet")
proto.RegisterType((*ManifestChange)(nil), "badgerpb2.ManifestChange")
proto.RegisterType((*BlockOffset)(nil), "badgerpb2.BlockOffset")
proto.RegisterType((*TableIndex)(nil), "badgerpb2.TableIndex")
proto.RegisterType((*Checksum)(nil), "badgerpb2.Checksum")
proto.RegisterType((*DataKey)(nil), "badgerpb2.DataKey")
}
@@ -650,51 +522,44 @@ func init() {
func init() { proto.RegisterFile("badgerpb2.proto", fileDescriptor_e63e84f9f0d3998c) }
var fileDescriptor_e63e84f9f0d3998c = []byte{
// 689 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x74, 0x54, 0xcd, 0x6e, 0xda, 0x40,
0x10, 0xc6, 0xc6, 0xe1, 0x67, 0x08, 0x84, 0xae, 0xda, 0xc8, 0x51, 0x15, 0x4a, 0x1c, 0x45, 0x45,
0x95, 0x0a, 0x2d, 0x54, 0xbd, 0x13, 0x42, 0x15, 0x44, 0x22, 0xa4, 0x4d, 0x14, 0x45, 0xbd, 0xa0,
0xc5, 0x1e, 0xc0, 0xc2, 0x7f, 0xf2, 0x2e, 0x56, 0xc8, 0x13, 0xf4, 0xd2, 0x7b, 0x1f, 0xa9, 0xc7,
0x1e, 0xfa, 0x00, 0x55, 0xfa, 0x22, 0x95, 0xd7, 0x86, 0x82, 0xd4, 0xde, 0x66, 0xbe, 0xf9, 0x76,
0x67, 0xbf, 0x6f, 0xc6, 0x86, 0x83, 0x09, 0xb3, 0x66, 0x18, 0x06, 0x93, 0x76, 0x33, 0x08, 0x7d,
0xe1, 0x93, 0xe2, 0x06, 0x30, 0x7e, 0x2a, 0xa0, 0x0e, 0xef, 0x48, 0x15, 0xb2, 0x0b, 0x5c, 0xe9,
0x4a, 0x5d, 0x69, 0xec, 0xd3, 0x38, 0x24, 0xcf, 0x61, 0x2f, 0x62, 0xce, 0x12, 0x75, 0x55, 0x62,
0x49, 0x42, 0x5e, 0x42, 0x71, 0xc9, 0x31, 0x1c, 0xbb, 0x28, 0x98, 0x9e, 0x95, 0x95, 0x42, 0x0c,
0x5c, 0xa3, 0x60, 0x44, 0x87, 0x7c, 0x84, 0x21, 0xb7, 0x7d, 0x4f, 0xd7, 0xea, 0x4a, 0x43, 0xa3,
0xeb, 0x94, 0x1c, 0x03, 0xe0, 0x43, 0x60, 0x87, 0xc8, 0xc7, 0x4c, 0xe8, 0x7b, 0xb2, 0x58, 0x4c,
0x91, 0xae, 0x20, 0x04, 0x34, 0x79, 0x61, 0x4e, 0x5e, 0x28, 0xe3, 0xb8, 0x13, 0x17, 0x21, 0x32,
0x77, 0x6c, 0x5b, 0x3a, 0xd4, 0x95, 0x46, 0x99, 0x16, 0x12, 0x60, 0x60, 0x91, 0x57, 0x50, 0x4a,
0x8b, 0x96, 0xef, 0xa1, 0x5e, 0xaa, 0x2b, 0x8d, 0x02, 0x85, 0x04, 0xba, 0xf0, 0x3d, 0x34, 0x5e,
0x43, 0x6e, 0x78, 0x77, 0x65, 0x73, 0x41, 0x8e, 0x41, 0x5d, 0x44, 0xba, 0x52, 0xcf, 0x36, 0x4a,
0xed, 0x72, 0xf3, 0xaf, 0x13, 0xc3, 0x3b, 0xaa, 0x2e, 0x22, 0xe3, 0x12, 0x9e, 0x5d, 0x33, 0xcf,
0x9e, 0x22, 0x17, 0xbd, 0x39, 0xf3, 0x66, 0x78, 0x83, 0x82, 0x74, 0x20, 0x6f, 0xca, 0x84, 0xa7,
0x07, 0x8f, 0xb6, 0x0e, 0xee, 0xd2, 0xe9, 0x9a, 0x69, 0x7c, 0x55, 0xa1, 0xb2, 0x5b, 0x23, 0x15,
0x50, 0x07, 0x96, 0x34, 0x55, 0xa3, 0xea, 0xc0, 0x22, 0x1d, 0x50, 0x47, 0x81, 0x34, 0xb4, 0xd2,
0x3e, 0xfd, 0xef, 0x95, 0xcd, 0x51, 0x80, 0x21, 0x13, 0xb6, 0xef, 0x51, 0x75, 0x14, 0xc4, 0x83,
0xb8, 0xc2, 0x08, 0x1d, 0x69, 0x77, 0x99, 0x26, 0x09, 0x79, 0x01, 0xb9, 0x05, 0xae, 0x62, 0x6f,
0x12, 0xab, 0xf7, 0x16, 0xb8, 0x1a, 0x58, 0xe4, 0x1c, 0x0e, 0xd0, 0x33, 0xc3, 0x55, 0x10, 0x1f,
0x1f, 0x33, 0x67, 0xe6, 0x4b, 0xb7, 0x2b, 0x3b, 0x0a, 0xfa, 0x1b, 0x46, 0xd7, 0x99, 0xf9, 0xb4,
0x82, 0x3b, 0x39, 0xa9, 0x43, 0xc9, 0xf4, 0xdd, 0x20, 0x44, 0x2e, 0x47, 0x99, 0x93, 0x6d, 0xb7,
0x21, 0xe3, 0x14, 0x8a, 0x9b, 0x37, 0x12, 0x80, 0x5c, 0x8f, 0xf6, 0xbb, 0xb7, 0xfd, 0x6a, 0x26,
0x8e, 0x2f, 0xfa, 0x57, 0xfd, 0xdb, 0x7e, 0x55, 0x31, 0x06, 0x50, 0x3a, 0x77, 0x7c, 0x73, 0x31,
0x9a, 0x4e, 0x39, 0x8a, 0x7f, 0x6c, 0xd8, 0x21, 0xe4, 0x7c, 0x59, 0x93, 0x8e, 0x94, 0x69, 0x9a,
0xc5, 0x4c, 0x07, 0xbd, 0x54, 0x6e, 0x1c, 0x1a, 0x5f, 0x14, 0x80, 0x5b, 0x36, 0x71, 0x70, 0xe0,
0x59, 0xf8, 0x40, 0xde, 0x41, 0x3e, 0xa1, 0xae, 0xc7, 0x73, 0xb8, 0x25, 0x6e, 0xab, 0x27, 0x5d,
0xd3, 0xc8, 0x09, 0xec, 0x4f, 0x1c, 0xdf, 0x77, 0xc7, 0x53, 0xdb, 0x11, 0x18, 0xa6, 0x3b, 0x5d,
0x92, 0xd8, 0x27, 0x09, 0x91, 0x33, 0xa8, 0x20, 0x17, 0xb6, 0xcb, 0x04, 0x5a, 0x63, 0x6e, 0x3f,
0xa2, 0x7c, 0x80, 0x46, 0xcb, 0x1b, 0xf4, 0xc6, 0x7e, 0x44, 0x23, 0x82, 0x42, 0x6f, 0x8e, 0xe6,
0x82, 0x2f, 0x5d, 0xf2, 0x1e, 0x34, 0xe9, 0xb0, 0x22, 0x1d, 0x3e, 0xde, 0x7a, 0xc4, 0x9a, 0xd2,
0x8c, 0x0d, 0x0d, 0x6d, 0x31, 0x77, 0xa9, 0xa4, 0xc6, 0xda, 0xf8, 0xd2, 0x95, 0xfd, 0x35, 0x1a,
0x87, 0xc6, 0x19, 0x14, 0x37, 0xa4, 0xc4, 0xcb, 0x5e, 0xa7, 0xdd, 0xab, 0x66, 0xc8, 0x3e, 0x14,
0xee, 0xef, 0x2f, 0x19, 0x9f, 0x7f, 0xfc, 0x50, 0x55, 0x0c, 0x13, 0xf2, 0x17, 0x4c, 0xb0, 0x21,
0xae, 0xb6, 0x46, 0xaf, 0x6c, 0x8f, 0x9e, 0x80, 0x66, 0x31, 0xc1, 0x52, 0x6d, 0x32, 0x8e, 0x17,
0xd0, 0x8e, 0xd2, 0xef, 0x54, 0xb5, 0xa3, 0xf8, 0x3b, 0x34, 0x43, 0x94, 0x12, 0x99, 0x90, 0x9b,
0x93, 0xa5, 0xc5, 0x14, 0xe9, 0x8a, 0x37, 0x47, 0x50, 0xd9, 0xdd, 0x0d, 0x92, 0x87, 0x2c, 0x43,
0x5e, 0xcd, 0x9c, 0x77, 0xbe, 0x3f, 0xd5, 0x94, 0x1f, 0x4f, 0x35, 0xe5, 0xd7, 0x53, 0x4d, 0xf9,
0xf6, 0xbb, 0x96, 0xf9, 0x7c, 0x32, 0xb3, 0xc5, 0x7c, 0x39, 0x69, 0x9a, 0xbe, 0xdb, 0xb2, 0x66,
0x21, 0x0b, 0xe6, 0x6f, 0x6d, 0xbf, 0x95, 0x78, 0xd0, 0x8a, 0xda, 0xad, 0x60, 0x32, 0xc9, 0xc9,
0xdf, 0x4d, 0xe7, 0x4f, 0x00, 0x00, 0x00, 0xff, 0xff, 0x88, 0x22, 0x82, 0x98, 0x81, 0x04, 0x00,
0x00,
// 589 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x74, 0x93, 0xcf, 0x6e, 0xda, 0x4e,
0x10, 0xc7, 0x59, 0xe3, 0xf0, 0x67, 0x48, 0x08, 0xbf, 0xd5, 0xaf, 0x92, 0xa3, 0x2a, 0x94, 0x3a,
0xaa, 0x8a, 0x2a, 0x15, 0x54, 0xa8, 0x7a, 0x27, 0x04, 0x29, 0x88, 0x44, 0x91, 0xb6, 0x51, 0x14,
0xf5, 0x82, 0x16, 0x7b, 0x0a, 0x16, 0xf8, 0x8f, 0x76, 0x17, 0xab, 0x3c, 0x44, 0xef, 0x7d, 0xa4,
0x1e, 0x7b, 0xe8, 0x03, 0x54, 0xe9, 0x8b, 0x54, 0xbb, 0x36, 0x14, 0x0e, 0xbd, 0xcd, 0x7c, 0xe7,
0xbb, 0x3b, 0xe3, 0xcf, 0x8e, 0xe1, 0x74, 0xc6, 0xfd, 0x39, 0x8a, 0x64, 0xd6, 0xeb, 0x24, 0x22,
0x56, 0x31, 0xad, 0xee, 0x04, 0xf7, 0x27, 0x01, 0x6b, 0xf2, 0x40, 0x1b, 0x50, 0x5c, 0xe2, 0xc6,
0x21, 0x2d, 0xd2, 0x3e, 0x66, 0x3a, 0xa4, 0xff, 0xc3, 0x51, 0xca, 0x57, 0x6b, 0x74, 0x2c, 0xa3,
0x65, 0x09, 0x7d, 0x0e, 0xd5, 0xb5, 0x44, 0x31, 0x0d, 0x51, 0x71, 0xa7, 0x68, 0x2a, 0x15, 0x2d,
0xdc, 0xa2, 0xe2, 0xd4, 0x81, 0x72, 0x8a, 0x42, 0x06, 0x71, 0xe4, 0xd8, 0x2d, 0xd2, 0xb6, 0xd9,
0x36, 0xa5, 0xe7, 0x00, 0xf8, 0x25, 0x09, 0x04, 0xca, 0x29, 0x57, 0xce, 0x91, 0x29, 0x56, 0x73,
0x65, 0xa0, 0x28, 0x05, 0xdb, 0x5c, 0x58, 0x32, 0x17, 0x9a, 0x58, 0x77, 0x92, 0x4a, 0x20, 0x0f,
0xa7, 0x81, 0xef, 0x40, 0x8b, 0xb4, 0x4f, 0x58, 0x25, 0x13, 0xc6, 0x3e, 0x7d, 0x01, 0xb5, 0xbc,
0xe8, 0xc7, 0x11, 0x3a, 0xb5, 0x16, 0x69, 0x57, 0x18, 0x64, 0xd2, 0x55, 0x1c, 0xa1, 0xfb, 0x1a,
0x4a, 0x93, 0x87, 0x9b, 0x40, 0x2a, 0x7a, 0x0e, 0xd6, 0x32, 0x75, 0x48, 0xab, 0xd8, 0xae, 0xf5,
0x4e, 0x3a, 0x7f, 0x49, 0x4c, 0x1e, 0x98, 0xb5, 0x4c, 0xdd, 0x6b, 0xf8, 0xef, 0x96, 0x47, 0xc1,
0x67, 0x94, 0x6a, 0xb8, 0xe0, 0xd1, 0x1c, 0x3f, 0xa2, 0xa2, 0x7d, 0x28, 0x7b, 0x26, 0x91, 0xf9,
0xc1, 0xb3, 0xbd, 0x83, 0x87, 0x76, 0xb6, 0x75, 0xba, 0x5f, 0x2d, 0xa8, 0x1f, 0xd6, 0x68, 0x1d,
0xac, 0xb1, 0x6f, 0xa0, 0xda, 0xcc, 0x1a, 0xfb, 0xb4, 0x0f, 0xd6, 0x5d, 0x62, 0x80, 0xd6, 0x7b,
0x17, 0xff, 0xbc, 0xb2, 0x73, 0x97, 0xa0, 0xe0, 0x2a, 0x88, 0x23, 0x66, 0xdd, 0x25, 0xfa, 0x21,
0x6e, 0x30, 0xc5, 0x95, 0xc1, 0x7d, 0xc2, 0xb2, 0x84, 0x3e, 0x83, 0xd2, 0x12, 0x37, 0x9a, 0x4d,
0x86, 0xfa, 0x68, 0x89, 0x9b, 0xb1, 0x4f, 0x2f, 0xe1, 0x14, 0x23, 0x4f, 0x6c, 0x12, 0x7d, 0x7c,
0xca, 0x57, 0xf3, 0xd8, 0xd0, 0xae, 0x1f, 0x7c, 0xc1, 0x68, 0xe7, 0x18, 0xac, 0xe6, 0x31, 0xab,
0xe3, 0x41, 0x4e, 0x5b, 0x50, 0xf3, 0xe2, 0x30, 0x11, 0x28, 0xcd, 0x53, 0x96, 0x4c, 0xdb, 0x7d,
0xc9, 0xbd, 0x80, 0xea, 0x6e, 0x46, 0x0a, 0x50, 0x1a, 0xb2, 0xd1, 0xe0, 0x7e, 0xd4, 0x28, 0xe8,
0xf8, 0x6a, 0x74, 0x33, 0xba, 0x1f, 0x35, 0x88, 0x9b, 0x42, 0x65, 0xb8, 0x40, 0x6f, 0x29, 0xd7,
0x21, 0x7d, 0x07, 0xb6, 0x99, 0x85, 0x98, 0x59, 0xce, 0xf7, 0x66, 0xd9, 0x5a, 0x3a, 0xba, 0xb5,
0x08, 0xd4, 0x22, 0x64, 0xc6, 0xaa, 0x37, 0x52, 0xae, 0x43, 0x03, 0xcb, 0x66, 0x3a, 0x74, 0x5f,
0x41, 0x75, 0x67, 0xca, 0xba, 0x0e, 0xfb, 0xbd, 0x61, 0xa3, 0x40, 0x8f, 0xa1, 0xf2, 0xf8, 0x78,
0xcd, 0xe5, 0xe2, 0xc3, 0xfb, 0x06, 0x71, 0x3d, 0x28, 0x5f, 0x71, 0xc5, 0x27, 0xb8, 0xd9, 0x83,
0x44, 0xf6, 0x21, 0x51, 0xb0, 0x7d, 0xae, 0x78, 0xbe, 0xd9, 0x26, 0xd6, 0x4f, 0x15, 0xa4, 0xf9,
0x46, 0x5b, 0x41, 0xaa, 0x37, 0xd6, 0x13, 0xc8, 0x15, 0xfa, 0x7a, 0x63, 0x35, 0xe3, 0x22, 0xab,
0xe6, 0xca, 0x40, 0xbd, 0x39, 0x83, 0xfa, 0x21, 0x45, 0x5a, 0x86, 0x22, 0x47, 0xd9, 0x28, 0x5c,
0xf6, 0xbf, 0x3f, 0x35, 0xc9, 0x8f, 0xa7, 0x26, 0xf9, 0xf5, 0xd4, 0x24, 0xdf, 0x7e, 0x37, 0x0b,
0x9f, 0x5e, 0xce, 0x03, 0xb5, 0x58, 0xcf, 0x3a, 0x5e, 0x1c, 0x76, 0xfd, 0xb9, 0xe0, 0xc9, 0xe2,
0x6d, 0x10, 0x77, 0x33, 0x06, 0xdd, 0xb4, 0xd7, 0x4d, 0x66, 0xb3, 0x92, 0xf9, 0x31, 0xfb, 0x7f,
0x02, 0x00, 0x00, 0xff, 0xff, 0x1d, 0xc3, 0x11, 0xa3, 0xab, 0x03, 0x00, 0x00,
}
func (m *KV) Marshal() (dAtA []byte, err error) {
@@ -916,103 +781,6 @@ func (m *ManifestChange) MarshalToSizedBuffer(dAtA []byte) (int, error) {
return len(dAtA) - i, nil
}
func (m *BlockOffset) Marshal() (dAtA []byte, err error) {
size := m.Size()
dAtA = make([]byte, size)
n, err := m.MarshalToSizedBuffer(dAtA[:size])
if err != nil {
return nil, err
}
return dAtA[:n], nil
}
func (m *BlockOffset) MarshalTo(dAtA []byte) (int, error) {
size := m.Size()
return m.MarshalToSizedBuffer(dAtA[:size])
}
func (m *BlockOffset) MarshalToSizedBuffer(dAtA []byte) (int, error) {
i := len(dAtA)
_ = i
var l int
_ = l
if m.XXX_unrecognized != nil {
i -= len(m.XXX_unrecognized)
copy(dAtA[i:], m.XXX_unrecognized)
}
if m.Len != 0 {
i = encodeVarintBadgerpb2(dAtA, i, uint64(m.Len))
i--
dAtA[i] = 0x18
}
if m.Offset != 0 {
i = encodeVarintBadgerpb2(dAtA, i, uint64(m.Offset))
i--
dAtA[i] = 0x10
}
if len(m.Key) > 0 {
i -= len(m.Key)
copy(dAtA[i:], m.Key)
i = encodeVarintBadgerpb2(dAtA, i, uint64(len(m.Key)))
i--
dAtA[i] = 0xa
}
return len(dAtA) - i, nil
}
func (m *TableIndex) Marshal() (dAtA []byte, err error) {
size := m.Size()
dAtA = make([]byte, size)
n, err := m.MarshalToSizedBuffer(dAtA[:size])
if err != nil {
return nil, err
}
return dAtA[:n], nil
}
func (m *TableIndex) MarshalTo(dAtA []byte) (int, error) {
size := m.Size()
return m.MarshalToSizedBuffer(dAtA[:size])
}
func (m *TableIndex) MarshalToSizedBuffer(dAtA []byte) (int, error) {
i := len(dAtA)
_ = i
var l int
_ = l
if m.XXX_unrecognized != nil {
i -= len(m.XXX_unrecognized)
copy(dAtA[i:], m.XXX_unrecognized)
}
if m.EstimatedSize != 0 {
i = encodeVarintBadgerpb2(dAtA, i, uint64(m.EstimatedSize))
i--
dAtA[i] = 0x18
}
if len(m.BloomFilter) > 0 {
i -= len(m.BloomFilter)
copy(dAtA[i:], m.BloomFilter)
i = encodeVarintBadgerpb2(dAtA, i, uint64(len(m.BloomFilter)))
i--
dAtA[i] = 0x12
}
if len(m.Offsets) > 0 {
for iNdEx := len(m.Offsets) - 1; iNdEx >= 0; iNdEx-- {
{
size, err := m.Offsets[iNdEx].MarshalToSizedBuffer(dAtA[:i])
if err != nil {
return 0, err
}
i -= size
i = encodeVarintBadgerpb2(dAtA, i, uint64(size))
}
i--
dAtA[i] = 0xa
}
}
return len(dAtA) - i, nil
}
func (m *Checksum) Marshal() (dAtA []byte, err error) {
size := m.Size()
dAtA = make([]byte, size)
@@ -1218,53 +986,6 @@ func (m *ManifestChange) Size() (n int) {
return n
}
func (m *BlockOffset) Size() (n int) {
if m == nil {
return 0
}
var l int
_ = l
l = len(m.Key)
if l > 0 {
n += 1 + l + sovBadgerpb2(uint64(l))
}
if m.Offset != 0 {
n += 1 + sovBadgerpb2(uint64(m.Offset))
}
if m.Len != 0 {
n += 1 + sovBadgerpb2(uint64(m.Len))
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
return n
}
func (m *TableIndex) Size() (n int) {
if m == nil {
return 0
}
var l int
_ = l
if len(m.Offsets) > 0 {
for _, e := range m.Offsets {
l = e.Size()
n += 1 + l + sovBadgerpb2(uint64(l))
}
}
l = len(m.BloomFilter)
if l > 0 {
n += 1 + l + sovBadgerpb2(uint64(l))
}
if m.EstimatedSize != 0 {
n += 1 + sovBadgerpb2(uint64(m.EstimatedSize))
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
return n
}
func (m *Checksum) Size() (n int) {
if m == nil {
return 0
@@ -1926,273 +1647,6 @@ func (m *ManifestChange) Unmarshal(dAtA []byte) error {
}
return nil
}
func (m *BlockOffset) Unmarshal(dAtA []byte) error {
l := len(dAtA)
iNdEx := 0
for iNdEx < l {
preIndex := iNdEx
var wire uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowBadgerpb2
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
wire |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
fieldNum := int32(wire >> 3)
wireType := int(wire & 0x7)
if wireType == 4 {
return fmt.Errorf("proto: BlockOffset: wiretype end group for non-group")
}
if fieldNum <= 0 {
return fmt.Errorf("proto: BlockOffset: illegal tag %d (wire type %d)", fieldNum, wire)
}
switch fieldNum {
case 1:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Key", wireType)
}
var byteLen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowBadgerpb2
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
byteLen |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
if byteLen < 0 {
return ErrInvalidLengthBadgerpb2
}
postIndex := iNdEx + byteLen
if postIndex < 0 {
return ErrInvalidLengthBadgerpb2
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Key = append(m.Key[:0], dAtA[iNdEx:postIndex]...)
if m.Key == nil {
m.Key = []byte{}
}
iNdEx = postIndex
case 2:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field Offset", wireType)
}
m.Offset = 0
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowBadgerpb2
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
m.Offset |= uint32(b&0x7F) << shift
if b < 0x80 {
break
}
}
case 3:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field Len", wireType)
}
m.Len = 0
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowBadgerpb2
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
m.Len |= uint32(b&0x7F) << shift
if b < 0x80 {
break
}
}
default:
iNdEx = preIndex
skippy, err := skipBadgerpb2(dAtA[iNdEx:])
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthBadgerpb2
}
if (iNdEx + skippy) < 0 {
return ErrInvalidLengthBadgerpb2
}
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
iNdEx += skippy
}
}
if iNdEx > l {
return io.ErrUnexpectedEOF
}
return nil
}
func (m *TableIndex) Unmarshal(dAtA []byte) error {
l := len(dAtA)
iNdEx := 0
for iNdEx < l {
preIndex := iNdEx
var wire uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowBadgerpb2
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
wire |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
fieldNum := int32(wire >> 3)
wireType := int(wire & 0x7)
if wireType == 4 {
return fmt.Errorf("proto: TableIndex: wiretype end group for non-group")
}
if fieldNum <= 0 {
return fmt.Errorf("proto: TableIndex: illegal tag %d (wire type %d)", fieldNum, wire)
}
switch fieldNum {
case 1:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Offsets", wireType)
}
var msglen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowBadgerpb2
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
msglen |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
if msglen < 0 {
return ErrInvalidLengthBadgerpb2
}
postIndex := iNdEx + msglen
if postIndex < 0 {
return ErrInvalidLengthBadgerpb2
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Offsets = append(m.Offsets, &BlockOffset{})
if err := m.Offsets[len(m.Offsets)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
return err
}
iNdEx = postIndex
case 2:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field BloomFilter", wireType)
}
var byteLen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowBadgerpb2
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
byteLen |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
if byteLen < 0 {
return ErrInvalidLengthBadgerpb2
}
postIndex := iNdEx + byteLen
if postIndex < 0 {
return ErrInvalidLengthBadgerpb2
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.BloomFilter = append(m.BloomFilter[:0], dAtA[iNdEx:postIndex]...)
if m.BloomFilter == nil {
m.BloomFilter = []byte{}
}
iNdEx = postIndex
case 3:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field EstimatedSize", wireType)
}
m.EstimatedSize = 0
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowBadgerpb2
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
m.EstimatedSize |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
default:
iNdEx = preIndex
skippy, err := skipBadgerpb2(dAtA[iNdEx:])
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthBadgerpb2
}
if (iNdEx + skippy) < 0 {
return ErrInvalidLengthBadgerpb2
}
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
iNdEx += skippy
}
}
if iNdEx > l {
return io.ErrUnexpectedEOF
}
return nil
}
func (m *Checksum) Unmarshal(dAtA []byte) error {
l := len(dAtA)
iNdEx := 0
pb/badgerpb2.proto
-12
View File
@@ -61,18 +61,6 @@ message ManifestChange {
uint32 compression = 6; // Only used for CREATE Op.
}
message BlockOffset {
bytes key = 1;
uint32 offset = 2;
uint32 len = 3;
}
message TableIndex {
repeated BlockOffset offsets = 1;
bytes bloom_filter = 2;
uint64 estimated_size = 3;
}
message Checksum {
enum Algorithm {
CRC32C = 0;
stream_writer_test.go
+1 -1
View File
@@ -344,7 +344,7 @@ func TestStreamWriter6(t *testing.T) {
require.NoError(t, sw.Write(list), "sw.Write() failed")
require.NoError(t, sw.Flush(), "sw.Flush() failed")
tables := db.Tables(true)
tables := db.Tables()
require.Equal(t, 4, len(tables), "Count of tables not matching")
for _, tab := range tables {
if tab.Level > 0 {
table/builder.go
+120 -39
View File
@@ -24,9 +24,10 @@ import (
"sync/atomic"
"unsafe"
"github.com/dgryski/go-farm"
"github.com/dgraph-io/badger/v2/fb"
"github.com/golang/protobuf/proto"
"github.com/golang/snappy"
fbs "github.com/google/flatbuffers/go"
"github.com/pkg/errors"
"github.com/dgraph-io/badger/v2/options"
@@ -66,6 +67,7 @@ func (h *header) Decode(buf []byte) {
copy(((*[headerSize]byte)(unsafe.Pointer(h))[:]), buf[:headerSize])
}
// bblock represents a block that is being compressed/encrypted in the background.
type bblock struct {
data []byte
start uint32 // Points to the starting offset of the block.
@@ -80,12 +82,15 @@ type Builder struct {
bufLock sync.Mutex // This lock guards the buf. We acquire lock when we resize the buf.
actualSize uint32 // Used to store the sum of sizes of blocks after compression/encryption.
baseKey []byte // Base key for the current block.
baseOffset uint32 // Offset for the current block.
entryOffsets []uint32 // Offsets of entries present in current block.
tableIndex *pb.TableIndex
keyHashes []uint64 // Used for building the bloomfilter.
opt *Options
baseKey []byte // Base key for the current block.
baseOffset uint32 // Offset for the current block.
entryOffsets []uint32 // Offsets of entries present in current block.
offsets *z.Buffer
estimatedSize uint32
keyHashes []uint32 // Used for building the bloomfilter.
opt *Options
maxVersion uint64
// Used to concurrently compress/encrypt blocks.
wg sync.WaitGroup
@@ -98,10 +103,9 @@ func NewTableBuilder(opts Options) *Builder {
b := &Builder{
// Additional 16 MB to store index (approximate).
// We trim the additional space in table.Finish().
buf: z.Calloc(int(opts.TableSize + 16*MB)),
tableIndex: &pb.TableIndex{},
keyHashes: make([]uint64, 0, 1024), // Avoid some malloc calls.
opt: &opts,
buf: z.Calloc(int(opts.TableSize + 16*MB)),
opt: &opts,
offsets: z.NewBuffer(1 << 20),
}
// If encryption or compression is not enabled, do not start compression/encryption goroutines
@@ -168,6 +172,7 @@ func (b *Builder) handleBlock() {
// Close closes the TableBuilder.
func (b *Builder) Close() {
b.offsets.Release()
z.Free(b.buf)
}
@@ -185,8 +190,12 @@ func (b *Builder) keyDiff(newKey []byte) []byte {
return newKey[i:]
}
func (b *Builder) addHelper(key []byte, v y.ValueStruct, vpLen uint64) {
b.keyHashes = append(b.keyHashes, farm.Fingerprint64(y.ParseKey(key)))
func (b *Builder) addHelper(key []byte, v y.ValueStruct, vpLen uint32) {
b.keyHashes = append(b.keyHashes, y.Hash(y.ParseKey(key)))
if version := y.ParseTs(key); version > b.maxVersion {
b.maxVersion = version
}
// diffKey stores the difference of key with baseKey.
var diffKey []byte
@@ -221,9 +230,9 @@ func (b *Builder) addHelper(key []byte, v y.ValueStruct, vpLen uint64) {
b.sz += v.Encode(b.buf[b.sz:])
// Size of KV on SST.
sstSz := uint64(uint32(headerSize) + uint32(len(diffKey)) + v.EncodedSize())
sstSz := uint32(headerSize) + uint32(len(diffKey)) + v.EncodedSize()
// Total estimated size = size on SST + size on vlog (length of value pointer).
b.tableIndex.EstimatedSize += (sstSz + vpLen)
b.estimatedSize += (sstSz + vpLen)
}
// grow increases the size of b.buf by atleast 50%.
@@ -297,12 +306,19 @@ func (b *Builder) finishBlock() {
func (b *Builder) addBlockToIndex() {
blockBuf := b.buf[b.baseOffset:b.sz]
// Add key to the block index.
bo := &pb.BlockOffset{
Key: y.Copy(b.baseKey),
Offset: b.baseOffset,
Len: uint32(len(blockBuf)),
}
b.tableIndex.Offsets = append(b.tableIndex.Offsets, bo)
builder := fbs.NewBuilder(64)
off := builder.CreateByteVector(b.baseKey)
fb.BlockOffsetStart(builder)
fb.BlockOffsetAddKey(builder, off)
fb.BlockOffsetAddOffset(builder, b.baseOffset)
fb.BlockOffsetAddLen(builder, uint32(len(blockBuf)))
uoff := fb.BlockOffsetEnd(builder)
builder.Finish(uoff)
out := builder.FinishedBytes()
dst := b.offsets.SliceAllocate(len(out))
copy(dst, out)
}
func (b *Builder) shouldFinishBlock(key []byte, value y.ValueStruct) bool {
@@ -342,7 +358,7 @@ func (b *Builder) Add(key []byte, value y.ValueStruct, valueLen uint32) {
b.baseOffset = uint32((b.sz))
b.entryOffsets = b.entryOffsets[:0]
}
b.addHelper(key, value, uint64(valueLen))
b.addHelper(key, value, valueLen)
}
// TODO: vvv this was the comment on ReachedCapacity.
@@ -358,9 +374,10 @@ func (b *Builder) ReachedCapacity(capacity uint64) bool {
4 + // count of all entry offsets
8 + // checksum bytes
4 // checksum length
estimateSz := blocksSize +
4 + // Index length
5*(uint32(len(b.tableIndex.Offsets))) // approximate index size
uint32(b.offsets.Len())
return uint64(estimateSz) > capacity
}
@@ -378,15 +395,6 @@ The table structure looks like
*/
// In case the data is encrypted, the "IV" is added to the end of the index.
func (b *Builder) Finish(allocate bool) []byte {
if b.opt.BloomFalsePositive > 0 {
bf := z.NewBloomFilter(float64(len(b.keyHashes)), b.opt.BloomFalsePositive)
for _, h := range b.keyHashes {
bf.Add(h)
}
// Add bloom filter to the index.
b.tableIndex.BloomFilter = bf.JSONMarshal()
}
b.finishBlock() // This will never start a new block.
if b.blockChan != nil {
@@ -395,32 +403,44 @@ func (b *Builder) Finish(allocate bool) []byte {
// Wait for block handler to finish.
b.wg.Wait()
// We have added padding after each block so we should minus the
// padding from the actual table size. len(blocklist) would be zero if
// there is no compression/encryption.
uncompressedSize := b.sz - uint32(padding*len(b.blockList))
dst := b.buf
// Fix block boundaries. This includes moving the blocks so that we
// don't have any interleaving space between them.
bo, next := []byte{}, 1
if len(b.blockList) > 0 {
dstLen := uint32(0)
for i, bl := range b.blockList {
off := b.tableIndex.Offsets[i]
for _, bl := range b.blockList {
bo, next = b.offsets.Slice(next)
// Length of the block is end minus the start.
off.Len = bl.end - bl.start
fbo := fb.GetRootAsBlockOffset(bo, 0)
fbo.MutateLen(bl.end - bl.start)
// New offset of the block is the point in the main buffer till
// which we have written data.
off.Offset = dstLen
fbo.MutateOffset(dstLen)
copy(dst[dstLen:], b.buf[bl.start:bl.end])
// New length is the start of the block plus its length.
dstLen = off.Offset + off.Len
dstLen = fbo.Offset() + fbo.Len()
}
y.AssertTrue(next == 0)
// Start writing to the buffer from the point until which we have valid data.
// Fix the length because append and writeChecksum also rely on it.
b.sz = dstLen
}
index, err := proto.Marshal(b.tableIndex)
y.Check(err)
var f y.Filter
if b.opt.BloomFalsePositive > 0 {
bits := y.BloomBitsPerKey(len(b.keyHashes), b.opt.BloomFalsePositive)
f = y.NewFilter(b.keyHashes, bits)
}
index := b.buildIndex(f, uncompressedSize)
var err error
if b.shouldEncrypt() {
index, err = b.encrypt(index, false)
y.Check(err)
@@ -518,3 +538,64 @@ func (b *Builder) compressData(data []byte) ([]byte, error) {
}
return nil, errors.New("Unsupported compression type")
}
func (b *Builder) buildIndex(bloom []byte, tableSz uint32) []byte {
builder := fbs.NewBuilder(3 << 20)
boList := b.writeBlockOffsets(builder)
// Write block offset vector the the idxBuilder.
fb.TableIndexStartOffsetsVector(builder, len(boList))
// Write individual block offsets.
for i := 0; i < len(boList); i++ {
builder.PrependUOffsetT(boList[i])
}
boEnd := builder.EndVector(len(boList))
var bfoff fbs.UOffsetT
// Write the bloom filter.
if len(bloom) > 0 {
bfoff = builder.CreateByteVector(bloom)
}
fb.TableIndexStart(builder)
fb.TableIndexAddOffsets(builder, boEnd)
fb.TableIndexAddBloomFilter(builder, bfoff)
fb.TableIndexAddEstimatedSize(builder, b.estimatedSize)
fb.TableIndexAddMaxVersion(builder, b.maxVersion)
fb.TableIndexAddUncompressedSize(builder, tableSz)
fb.TableIndexAddKeyCount(builder, uint32(len(b.keyHashes)))
builder.Finish(fb.TableIndexEnd(builder))
return builder.FinishedBytes()
}
// writeBlockOffsets writes all the blockOffets in b.offsets and returns the
// offsets for the newly written items.
func (b *Builder) writeBlockOffsets(builder *fbs.Builder) []fbs.UOffsetT {
so := b.offsets.SliceOffsets()
var uoffs []fbs.UOffsetT
for i := len(so) - 1; i >= 0; i-- {
// We add these in reverse order.
data, _ := b.offsets.Slice(so[i])
uoff := b.writeBlockOffset(builder, data)
uoffs = append(uoffs, uoff)
}
return uoffs
}
// writeBlockOffset writes the given key,offset,len triple to the indexBuilder.
// It returns the offset of the newly written blockoffset.
func (b *Builder) writeBlockOffset(builder *fbs.Builder, data []byte) fbs.UOffsetT {
// Write the key to the buffer.
bo := fb.GetRootAsBlockOffset(data, 0)
k := builder.CreateByteVector(bo.KeyBytes())
// Build the blockOffset.
fb.BlockOffsetStart(builder)
fb.BlockOffsetAddKey(builder, k)
fb.BlockOffsetAddOffset(builder, bo.Offset())
fb.BlockOffsetAddLen(builder, bo.Len())
return fb.BlockOffsetEnd(builder)
}
table/builder_test.go
+11 -8
View File
@@ -23,9 +23,9 @@ import (
"testing"
"time"
"github.com/dgryski/go-farm"
"github.com/stretchr/testify/require"
"github.com/dgraph-io/badger/v2/fb"
"github.com/dgraph-io/badger/v2/options"
"github.com/dgraph-io/badger/v2/pb"
"github.com/dgraph-io/badger/v2/y"
@@ -95,7 +95,7 @@ func TestTableIndex(t *testing.T) {
blockFirstKeys := make([][]byte, 0)
blockCount := 0
for i := 0; i < keysCount; i++ {
k := []byte(fmt.Sprintf("%016x", i))
k := y.KeyWithTs([]byte(fmt.Sprintf("%016x", i)), uint64(i+1))
v := fmt.Sprintf("%d", i)
vs := y.ValueStruct{Value: []byte(v)}
if i == 0 { // This is first key for first block.
@@ -119,12 +119,15 @@ func TestTableIndex(t *testing.T) {
}
// Ensure index is built correctly
require.Equal(t, blockCount, tbl.noOfBlocks)
require.Equal(t, blockCount, tbl.offsetsLength())
idx, err := tbl.readTableIndex()
require.NoError(t, err)
for i, ko := range idx.Offsets {
require.Equal(t, ko.Key, blockFirstKeys[i])
for i := 0; i < idx.OffsetsLength(); i++ {
var bo fb.BlockOffset
require.True(t, idx.Offsets(&bo, i))
require.Equal(t, blockFirstKeys[i], bo.KeyBytes())
}
require.Equal(t, keysCount, int(tbl.MaxVersion()))
f.Close()
require.NoError(t, os.RemoveAll(filename))
})
@@ -232,7 +235,7 @@ func TestBloomfilter(t *testing.T) {
c := 0
for it.Rewind(); it.Valid(); it.Next() {
c++
hash := farm.Fingerprint64(y.ParseKey(it.Key()))
hash := y.Hash(y.ParseKey(it.Key()))
require.False(t, tab.DoesNotHave(hash))
}
require.Equal(t, keyCount, c)
@@ -242,13 +245,13 @@ func TestBloomfilter(t *testing.T) {
c = 0
for it.Rewind(); it.Valid(); it.Next() {
c++
hash := farm.Fingerprint64(y.ParseKey(it.Key()))
hash := y.Hash(y.ParseKey(it.Key()))
require.False(t, tab.DoesNotHave(hash))
}
require.Equal(t, keyCount, c)
// Ensure tab.DoesNotHave works
hash := farm.Fingerprint64([]byte("foo"))
hash := y.Hash([]byte("foo"))
require.Equal(t, withBlooms, tab.DoesNotHave(hash))
}
table/iterator.go
+10 -7
View File
@@ -21,6 +21,7 @@ import (
"io"
"sort"
"github.com/dgraph-io/badger/v2/fb"
"github.com/dgraph-io/badger/v2/y"
"github.com/pkg/errors"
)
@@ -198,7 +199,7 @@ func (itr *Iterator) useCache() bool {
}
func (itr *Iterator) seekToFirst() {
numBlocks := itr.t.noOfBlocks
numBlocks := itr.t.offsetsLength()
if numBlocks == 0 {
itr.err = io.EOF
return
@@ -215,7 +216,7 @@ func (itr *Iterator) seekToFirst() {
}
func (itr *Iterator) seekToLast() {
numBlocks := itr.t.noOfBlocks
numBlocks := itr.t.offsetsLength()
if numBlocks == 0 {
itr.err = io.EOF
return
@@ -252,9 +253,11 @@ func (itr *Iterator) seekFrom(key []byte, whence int) {
case current:
}
idx := sort.Search(itr.t.noOfBlocks, func(idx int) bool {
ko := itr.t.blockOffsets()[idx]
return y.CompareKeys(ko.Key, key) > 0
var ko fb.BlockOffset
idx := sort.Search(itr.t.offsetsLength(), func(idx int) bool {
// Offsets should never return false since we're iterating within the OffsetsLength.
y.AssertTrue(itr.t.offsets(&ko, idx))
return y.CompareKeys(ko.KeyBytes(), key) > 0
})
if idx == 0 {
// The smallest key in our table is already strictly > key. We can return that.
@@ -272,7 +275,7 @@ func (itr *Iterator) seekFrom(key []byte, whence int) {
itr.seekHelper(idx-1, key)
if itr.err == io.EOF {
// Case 1. Need to visit block[idx].
if idx == itr.t.noOfBlocks {
if idx == itr.t.offsetsLength() {
// If idx == len(itr.t.blockIndex), then input key is greater than ANY element of table.
// There's nothing we can do. Valid() should return false as we seek to end of table.
return
@@ -300,7 +303,7 @@ func (itr *Iterator) seekForPrev(key []byte) {
func (itr *Iterator) next() {
itr.err = nil
if itr.bpos >= itr.t.noOfBlocks {
if itr.bpos >= itr.t.offsetsLength() {
itr.err = io.EOF
return
}
table/table.go
+88 -144
View File
@@ -36,6 +36,7 @@ import (
"github.com/golang/snappy"
"github.com/pkg/errors"
"github.com/dgraph-io/badger/v2/fb"
"github.com/dgraph-io/badger/v2/options"
"github.com/dgraph-io/badger/v2/pb"
"github.com/dgraph-io/badger/v2/y"
@@ -46,13 +47,6 @@ import (
const fileSuffix = ".sst"
const intSize = int(unsafe.Sizeof(int(0)))
// 1 word = 8 bytes
// sizeOfOffsetStruct is the size of pb.BlockOffset
const sizeOfOffsetStruct int64 = 3*8 + // key array take 3 words
1*8 + // offset and len takes 1 word
3*8 + // XXX_unrecognized array takes 3 word.
1*8 // so far 7 words, in order to round the slab we're adding one more word.
// Options contains configurable options for Table/Builder.
type Options struct {
// Options for Opening/Building Table.
@@ -80,6 +74,7 @@ type Options struct {
// Compression indicates the compression algorithm used for block compression.
Compression options.CompressionType
// Block cache is used to cache decompressed and decrypted blocks.
BlockCache *ristretto.Cache
IndexCache *ristretto.Cache
@@ -95,7 +90,7 @@ type Options struct {
type TableInterface interface {
Smallest() []byte
Biggest() []byte
DoesNotHave(hash uint64) bool
DoesNotHave(hash uint32) bool
}
// Table represents a loaded table file with the info we have about it.
@@ -104,11 +99,9 @@ type Table struct {
fd *os.File // Own fd.
tableSize int // Initialized in OpenTable, using fd.Stat().
bfLock sync.Mutex
blockOffset []*pb.BlockOffset
ref int32 // For file garbage collection. Atomic.
bf *z.Bloom // Nil if index cache in enabled.
index *fb.TableIndex // Nil if encryption is enabled.
ref int32 // For file garbage collection. Atomic.
mmap []byte // Memory mapped.
@@ -118,15 +111,18 @@ type Table struct {
Checksum []byte
// Stores the total size of key-values stored in this table (including the size on vlog).
estimatedSize uint64
estimatedSize uint32
indexStart int
indexLen int
hasBloomFilter bool
IsInmemory bool // Set to true if the table is on level 0 and opened in memory.
opt *Options
}
noOfBlocks int // Total number of blocks.
// MaxVersion returns the maximum version across all keys stored in this table.
func (t *Table) MaxVersion() uint64 {
return t.fetchIndex().MaxVersion()
}
// CompressionType returns the compression algorithm used for block compression.
@@ -146,6 +142,11 @@ func (t *Table) DecrRef() error {
// We can safely delete this file, because for all the current files, we always have
// at least one reference pointing to them.
// Delete all blocks from the cache.
for i := 0; i < t.offsetsLength(); i++ {
t.opt.BlockCache.Del(t.blockCacheKey(i))
}
// It's necessary to delete windows files.
if t.opt.LoadingMode == options.MemoryMap {
if err := y.Munmap(t.mmap); err != nil {
@@ -169,13 +170,6 @@ func (t *Table) DecrRef() error {
if err := os.Remove(filename); err != nil {
return err
}
// Delete all blocks from the cache.
for i := 0; i < t.noOfBlocks; i++ {
t.opt.BlockCache.Del(t.blockCacheKey(i))
}
// Delete bloom filter and indices from the cache.
t.opt.IndexCache.Del(t.blockOffsetsCacheKey())
t.opt.IndexCache.Del(t.bfCacheKey())
}
return nil
}
@@ -355,12 +349,12 @@ func OpenInMemoryTable(data []byte, id uint64, opt *Options) (*Table, error) {
func (t *Table) initBiggestAndSmallest() error {
var err error
var ko *pb.BlockOffset
var ko *fb.BlockOffset
if ko, err = t.initIndex(); err != nil {
return errors.Wrapf(err, "failed to read index.")
}
t.smallest = ko.Key
t.smallest = ko.KeyBytes()
it2 := t.NewIterator(REVERSED | NOCACHE)
defer it2.Close()
@@ -372,7 +366,7 @@ func (t *Table) initBiggestAndSmallest() error {
return nil
}
// Close closes the open table. (Releases resources back to the OS.)
// Close closes the open table. (Releases resources back to the OS.)
func (t *Table) Close() error {
if t.opt.LoadingMode == options.MemoryMap {
if err := y.Munmap(t.mmap); err != nil {
@@ -409,7 +403,7 @@ func (t *Table) readNoFail(off, sz int) []byte {
// initIndex reads the index and populate the necessary table fields and returns
// first block offset
func (t *Table) initIndex() (*pb.BlockOffset, error) {
func (t *Table) initIndex() (*fb.BlockOffset, error) {
readPos := t.tableSize
// Read checksum len from the last 4 bytes.
@@ -448,36 +442,22 @@ func (t *Table) initIndex() (*pb.BlockOffset, error) {
}
if t.opt.Compression == options.None {
t.estimatedSize = index.EstimatedSize
t.estimatedSize = index.EstimatedSize()
} else {
// Due to compression the real size on disk is much
// smaller than what we estimate from index.EstimatedSize.
t.estimatedSize = uint64(t.tableSize)
t.estimatedSize = uint32(t.tableSize)
}
t.hasBloomFilter = len(index.BloomFilter) > 0
t.noOfBlocks = len(index.Offsets)
t.hasBloomFilter = len(index.BloomFilterBytes()) > 0
// No cache
if t.opt.IndexCache == nil {
// Keep blooms in memory.
if t.hasBloomFilter && t.opt.LoadBloomsOnOpen {
bf, err := z.JSONUnmarshal(index.BloomFilter)
if err != nil {
return nil,
errors.Wrapf(err, "failed to unmarshal bloomfilter for table:%d", t.id)
}
t.bfLock.Lock()
t.bf = bf
t.bfLock.Unlock()
}
// Keep block offsets in memory since there is no cache.
t.blockOffset = index.Offsets
// If there's no encryption, this points to the mmap'ed buffer.
t.index = index
}
// We don't need to put anything in the indexCache here. Table.Open will
// create an iterator and that iterator will push the indices in cache.
return index.Offsets[0], nil
var bo fb.BlockOffset
y.AssertTrue(index.Offsets(&bo, 0))
return &bo, nil
}
// KeySplits splits the table into at least n ranges based on the block offsets.
@@ -486,56 +466,47 @@ func (t *Table) KeySplits(n int, prefix []byte) []string {
return nil
}
var res []string
offsets := t.blockOffsets()
jump := len(offsets) / n
oLen := t.offsetsLength()
jump := oLen / n
if jump == 0 {
jump = 1
}
for i := 0; i < len(offsets); i += jump {
if i >= len(offsets) {
i = len(offsets) - 1
var bo fb.BlockOffset
var res []string
for i := 0; i < oLen; i += jump {
if i >= oLen {
i = oLen - 1
}
if bytes.HasPrefix(offsets[i].Key, prefix) {
res = append(res, string(offsets[i].Key))
y.AssertTrue(t.offsets(&bo, i))
if bytes.HasPrefix(bo.KeyBytes(), prefix) {
res = append(res, string(bo.KeyBytes()))
}
}
return res
}
// blockOffsets returns block offsets of this table.
func (t *Table) blockOffsets() []*pb.BlockOffset {
func (t *Table) fetchIndex() *fb.TableIndex {
if t.opt.IndexCache == nil {
return t.blockOffset
return t.index
}
if val, ok := t.opt.IndexCache.Get(t.blockOffsetsCacheKey()); ok && val != nil {
return val.([]*pb.BlockOffset)
if val, ok := t.opt.IndexCache.Get(t.indexKey()); ok && val != nil {
return val.(*fb.TableIndex)
}
index, err := t.readTableIndex()
y.Check(err)
t.opt.IndexCache.Set(
t.blockOffsetsCacheKey(),
index.Offsets,
calculateOffsetsSize(index.Offsets))
return index.Offsets
t.opt.IndexCache.Set(t.indexKey(), index, int64(t.indexLen))
return index
}
// calculateOffsetsSize returns the size of *pb.BlockOffset array
func calculateOffsetsSize(offsets []*pb.BlockOffset) int64 {
totalSize := sizeOfOffsetStruct * int64(len(offsets))
func (t *Table) offsetsLength() int {
return t.fetchIndex().OffsetsLength()
}
for _, ko := range offsets {
// add key size.
totalSize += int64(cap(ko.Key))
// add XXX_unrecognized size.
totalSize += int64(cap(ko.XXX_unrecognized))
}
// Add three words for array size.
return totalSize + 3*8
func (t *Table) offsets(ko *fb.BlockOffset, i int) bool {
return t.fetchIndex().Offsets(ko, i)
}
// block function return a new block. Each block holds a ref and the byte
@@ -543,7 +514,7 @@ func calculateOffsetsSize(offsets []*pb.BlockOffset) int64 {
// caller should release the block by calling block.decrRef() on it.
func (t *Table) block(idx int, useCache bool) (*block, error) {
y.AssertTruef(idx >= 0, "idx=%d", idx)
if idx >= t.noOfBlocks {
if idx >= t.offsetsLength() {
return nil, errors.New("block out of index")
}
if t.opt.BlockCache != nil {
@@ -559,19 +530,20 @@ func (t *Table) block(idx int, useCache bool) (*block, error) {
}
}
// Read the block index if it's nil
ko := t.blockOffsets()[idx]
var ko fb.BlockOffset
y.AssertTrue(t.offsets(&ko, idx))
blk := &block{
offset: int(ko.Offset),
offset: int(ko.Offset()),
ref: 1,
}
defer blk.decrRef() // Deal with any errors, where blk would not be returned.
atomic.AddInt32(&NumBlocks, 1)
var err error
if blk.data, err = t.read(blk.offset, int(ko.Len)); err != nil {
if blk.data, err = t.read(blk.offset, int(ko.Len())); err != nil {
return nil, errors.Wrapf(err,
"failed to read from file: %s at offset: %d, len: %d", t.fd.Name(), blk.offset, ko.Len)
"failed to read from file: %s at offset: %d, len: %d",
t.fd.Name(), blk.offset, ko.Len())
}
if t.shouldDecrypt() {
@@ -584,7 +556,7 @@ func (t *Table) block(idx int, useCache bool) (*block, error) {
if err = t.decompress(blk); err != nil {
return nil, errors.Wrapf(err,
"failed to decode compressed data in file: %s at offset: %d, len: %d",
t.fd.Name(), blk.offset, ko.Len)
t.fd.Name(), blk.offset, ko.Len())
}
// Read meta data related to block.
@@ -639,18 +611,6 @@ func (t *Table) block(idx int, useCache bool) (*block, error) {
return blk, nil
}
// bfCacheKey returns the cache key for bloom filter. Bloom filters are stored in index cache.
func (t *Table) bfCacheKey() []byte {
y.AssertTrue(t.id < math.MaxUint32)
buf := make([]byte, 6)
// Without the "bf" prefix, we will have conflict with the blockCacheKey.
buf[0] = 'b'
buf[1] = 'f'
binary.BigEndian.PutUint32(buf[2:], uint32(t.id))
return buf
}
// blockCacheKey is used to store blocks in the block cache.
func (t *Table) blockCacheKey(idx int) []byte {
y.AssertTrue(t.id < math.MaxUint32)
@@ -663,31 +623,35 @@ func (t *Table) blockCacheKey(idx int) []byte {
return buf
}
// blockOffsetsCacheKey returns the cache key for block offsets. blockOffsets
// indexKey returns the cache key for block offsets. blockOffsets
// are stored in the index cache.
func (t *Table) blockOffsetsCacheKey() uint64 {
func (t *Table) indexKey() uint64 {
return t.id
}
// IndexSize returns the size of table index in bytes stored in the memory. The
// size of on-disk representation would be less than the in-memory representation.
func (t *Table) IndexSize() int {
indexSz := 0
for _, bi := range t.blockOffsets() {
indexSz += bi.Size()
}
return indexSz
// UncompressedSize is the size of table index in bytes.
func (t *Table) UncompressedSize() uint32 {
return t.index.UncompressedSize()
}
// BloomFilterSize returns the size of the bloom filter in bytes stored in memory. The
// size of on-disk representation would be less than the in-memory representation.
// KeyCount is the total number of keys in this table.
func (t *Table) KeyCount() uint32 {
return t.index.KeyCount()
}
// IndexSize is the size of table index in bytes.
func (t *Table) IndexSize() int {
return t.indexLen
}
// BloomFilterSize returns the size of the bloom filter in bytes stored in memory.
func (t *Table) BloomFilterSize() int {
return t.bf.TotalSize()
return t.index.BloomFilterLength()
}
// EstimatedSize returns the total size of key-values stored in this table (including the
// disk space occupied on the value log).
func (t *Table) EstimatedSize() uint64 { return t.estimatedSize }
func (t *Table) EstimatedSize() uint32 { return t.estimatedSize }
// Size is its file size in bytes
func (t *Table) Size() int64 { return int64(t.tableSize) }
@@ -706,49 +670,29 @@ func (t *Table) ID() uint64 { return t.id }
// DoesNotHave returns true if and only if the table does not have the key hash.
// It does a bloom filter lookup.
func (t *Table) DoesNotHave(hash uint64) bool {
func (t *Table) DoesNotHave(hash uint32) bool {
if !t.hasBloomFilter {
return false
}
// Return fast if the cache is absent.
if t.opt.IndexCache == nil {
t.bfLock.Lock()
if t.bf == nil {
y.AssertTrue(!t.opt.LoadBloomsOnOpen)
// Load bloomfilter into memory since the cache is absent.
t.bf, _ = t.readBloomFilter()
}
t.bfLock.Unlock()
return !t.bf.Has(hash)
}
// Check if the bloom filter exists in the cache.
if bf, ok := t.opt.IndexCache.Get(t.bfCacheKey()); bf != nil && ok {
return !bf.(*z.Bloom).Has(hash)
}
bf, sz := t.readBloomFilter()
t.opt.IndexCache.Set(t.bfCacheKey(), bf, int64(sz))
return !bf.Has(hash)
index := t.fetchIndex()
bf := index.BloomFilterBytes()
return !y.Filter(bf).MayContain(hash)
}
// readBloomFilter reads the bloom filter from the SST and returns its length
// along with the bloom filter.
func (t *Table) readBloomFilter() (*z.Bloom, int) {
// Read bloom filter from the SST.
index, err := t.readTableIndex()
index := t.fetchIndex()
// Read bloom filter from the index.
bf, err := z.JSONUnmarshal(index.BloomFilterBytes())
y.Check(err)
bf, err := z.JSONUnmarshal(index.BloomFilter)
y.Check(err)
return bf, len(index.BloomFilter)
return bf, index.BloomFilterLength()
}
// readTableIndex reads table index from the sst and returns its pb format.
func (t *Table) readTableIndex() (*pb.TableIndex, error) {
func (t *Table) readTableIndex() (*fb.TableIndex, error) {
data := t.readNoFail(t.indexStart, t.indexLen)
index := pb.TableIndex{}
var err error
// Decrypt the table index if it is encrypted.
if t.shouldDecrypt() {
@@ -757,18 +701,18 @@ func (t *Table) readTableIndex() (*pb.TableIndex, error) {
"Error while decrypting table index for the table %d in readTableIndex", t.id)
}
}
y.Check(proto.Unmarshal(data, &index))
return &index, nil
return fb.GetRootAsTableIndex(data, 0), nil
}
// VerifyChecksum verifies checksum for all blocks of table. This function is called by
// OpenTable() function. This function is also called inside levelsController.VerifyChecksum().
func (t *Table) VerifyChecksum() error {
for i, os := range t.blockOffsets() {
ti := t.fetchIndex()
for i := 0; i < ti.OffsetsLength(); i++ {
b, err := t.block(i, true)
if err != nil {
return y.Wrapf(err, "checksum validation failed for table: %s, block: %d, offset:%d",
t.Filename(), i, os.Offset)
t.Filename(), i, b.offset)
}
// We should not call incrRef here, because the block already has one ref when created.
defer b.decrRef()
@@ -778,7 +722,7 @@ func (t *Table) VerifyChecksum() error {
if err = b.verifyCheckSum(); err != nil {
return y.Wrapf(err,
"checksum validation failed for table: %s, block: %d, offset:%d",
t.Filename(), i, os.Offset)
t.Filename(), i, b.offset)
}
}
}
table/table_test.go
+31 -25
View File
@@ -31,7 +31,6 @@ import (
"github.com/cespare/xxhash"
"github.com/dgraph-io/badger/v2/options"
"github.com/dgraph-io/badger/v2/pb"
"github.com/dgraph-io/badger/v2/y"
"github.com/dgraph-io/ristretto"
"github.com/stretchr/testify/require"
@@ -80,7 +79,8 @@ func buildTable(t *testing.T, keyValues [][]string, opts Options) *os.File {
})
for _, kv := range keyValues {
y.AssertTrue(len(kv) == 2)
b.Add(y.KeyWithTs([]byte(kv[0]), 0), y.ValueStruct{Value: []byte(kv[1]), Meta: 'A', UserMeta: 0}, 0)
b.Add(y.KeyWithTs([]byte(kv[0]), 0),
y.ValueStruct{Value: []byte(kv[1]), Meta: 'A', UserMeta: 0}, 0)
}
_, err = f.Write(b.Finish(false))
require.NoError(t, err, "writing to file failed")
@@ -699,7 +699,7 @@ func TestTableBigValues(t *testing.T) {
TableSize: uint64(n) * 1 << 20}
builder := NewTableBuilder(opts)
for i := 0; i < n; i++ {
key := y.KeyWithTs([]byte(key("", i)), 0)
key := y.KeyWithTs([]byte(key("", i)), uint64(i+1))
vs := y.ValueStruct{Value: value(i)}
builder.Add(key, vs, 0)
}
@@ -721,6 +721,7 @@ func TestTableBigValues(t *testing.T) {
}
require.False(t, itr.Valid(), "table iterator should be invalid now")
require.Equal(t, n, count)
require.Equal(t, n, int(tbl.MaxVersion()))
}
// This test is for verifying checksum failure during table open.
@@ -739,10 +740,13 @@ func TestTableChecksum(t *testing.T) {
require.NoError(t, err)
require.Equal(t, n, len(rb))
_, err = OpenTable(f, opts)
if err == nil || !strings.Contains(err.Error(), "checksum") {
t.Fatal("Test should have been failed with checksum mismatch error")
}
require.Panics(t, func() {
// Either OpenTable will panic on corrupted data or the checksum verification will fail.
_, err = OpenTable(f, opts)
if strings.Contains(err.Error(), "checksum") {
panic("checksum mismatch")
}
})
}
var cacheConfig = ristretto.Config{
@@ -932,7 +936,7 @@ func TestOpenKVSize(t *testing.T) {
require.NoError(t, err)
// The estimated size is same as table size in case compression is enabled.
require.Equal(t, uint64(table.tableSize), table.EstimatedSize())
require.Equal(t, uint32(table.tableSize), table.EstimatedSize())
})
t.Run("no compressin", func(t *testing.T) {
@@ -945,7 +949,7 @@ func TestOpenKVSize(t *testing.T) {
stat, err := table.fd.Stat()
require.NoError(t, err)
require.Less(t, table.EstimatedSize(), uint64(stat.Size()))
require.Less(t, table.EstimatedSize(), uint32(stat.Size()))
})
}
@@ -961,28 +965,30 @@ func TestDoesNotHaveRace(t *testing.T) {
wg.Add(5)
for i := 0; i < 5; i++ {
go func() {
require.True(t, table.DoesNotHave(uint64(1237882)))
require.True(t, table.DoesNotHave(uint32(1237882)))
wg.Done()
}()
}
wg.Wait()
}
var ko *pb.BlockOffset
func TestMaxVersion(t *testing.T) {
opt := getTestTableOptions()
b := NewTableBuilder(opt)
defer b.Close()
// Use this benchmark to manually verify block offset size calculation
func BenchmarkBlockOffsetSizeCalculation(b *testing.B) {
for i := 0; i < b.N; i++ {
ko = &pb.BlockOffset{
Key: []byte{1, 23},
}
filename := fmt.Sprintf("%s%s%d.sst", os.TempDir(), string(os.PathSeparator), rand.Uint32())
f, err := y.CreateSyncedFile(filename, true)
require.NoError(t, err)
N := 1000
for i := 0; i < N; i++ {
b.Add(y.KeyWithTs([]byte(fmt.Sprintf("foo:%d", i)), uint64(i+1)), y.ValueStruct{}, 0)
}
}
func TestBlockOffsetSizeCalculation(t *testing.T) {
// Empty struct testing.
require.Equal(t, calculateOffsetsSize([]*pb.BlockOffset{&pb.BlockOffset{}}), int64(88))
// Testing with key bytes
require.Equal(t, calculateOffsetsSize([]*pb.BlockOffset{&pb.BlockOffset{Key: []byte{1, 1}}}),
int64(90))
_, err = f.Write(b.Finish(false))
require.NoError(t, err, "writing to file failed")
f.Close()
f, _ = y.OpenSyncedFile(filename, false)
table, err := OpenTable(f, opt)
require.Equal(t, N, int(table.MaxVersion()))
}
y/bloom.go
+170
View File
@@ -0,0 +1,170 @@
// Copyright 2013 The LevelDB-Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package y
import "math"
// Filter is an encoded set of []byte keys.
type Filter []byte
func (f Filter) MayContainKey(k []byte) bool {
return f.MayContain(Hash(k))
}
// MayContain returns whether the filter may contain given key. False positives
// are possible, where it returns true for keys not in the original set.
func (f Filter) MayContain(h uint32) bool {
if len(f) < 2 {
return false
}
k := f[len(f)-1]
if k > 30 {
// This is reserved for potentially new encodings for short Bloom filters.
// Consider it a match.
return true
}
nBits := uint32(8 * (len(f) - 1))
delta := h>>17 | h<<15
for j := uint8(0); j < k; j++ {
bitPos := h % nBits
if f[bitPos/8]&(1<<(bitPos%8)) == 0 {
return false
}
h += delta
}
return true
}
// NewFilter returns a new Bloom filter that encodes a set of []byte keys with
// the given number of bits per key, approximately.
//
// A good bitsPerKey value is 10, which yields a filter with ~ 1% false
// positive rate.
func NewFilter(keys []uint32, bitsPerKey int) Filter {
return Filter(appendFilter(nil, keys, bitsPerKey))
}
// BloomBitsPerKey returns the bits per key required by bloomfilter based on
// the false positive rate.
func BloomBitsPerKey(numEntries int, fp float64) int {
size := -1 * float64(numEntries) * math.Log(fp) / math.Pow(float64(0.69314718056), 2)
locs := math.Ceil(float64(0.69314718056) * size / float64(numEntries))
return int(locs)
}
func appendFilter(buf []byte, keys []uint32, bitsPerKey int) []byte {
if bitsPerKey < 0 {
bitsPerKey = 0
}
// 0.69 is approximately ln(2).
k := uint32(float64(bitsPerKey) * 0.69)
if k < 1 {
k = 1
}
if k > 30 {
k = 30
}
nBits := len(keys) * int(bitsPerKey)
// For small len(keys), we can see a very high false positive rate. Fix it
// by enforcing a minimum bloom filter length.
if nBits < 64 {
nBits = 64
}
nBytes := (nBits + 7) / 8
nBits = nBytes * 8
buf, filter := extend(buf, nBytes+1)
for _, h := range keys {
delta := h>>17 | h<<15
for j := uint32(0); j < k; j++ {
bitPos := h % uint32(nBits)
filter[bitPos/8] |= 1 << (bitPos % 8)
h += delta
}
}
filter[nBytes] = uint8(k)
return buf
}
// extend appends n zero bytes to b. It returns the overall slice (of length
// n+len(originalB)) and the slice of n trailing zeroes.
func extend(b []byte, n int) (overall, trailer []byte) {
want := n + len(b)
if want <= cap(b) {
overall = b[:want]
trailer = overall[len(b):]
for i := range trailer {
trailer[i] = 0
}
} else {
// Grow the capacity exponentially, with a 1KiB minimum.
c := 1024
for c < want {
c += c / 4
}
overall = make([]byte, want, c)
trailer = overall[len(b):]
copy(overall, b)
}
return overall, trailer
}
// hash implements a hashing algorithm similar to the Murmur hash.
func Hash(b []byte) uint32 {
const (
seed = 0xbc9f1d34
m = 0xc6a4a793
)
h := uint32(seed) ^ uint32(len(b)*m)
for ; len(b) >= 4; b = b[4:] {
h += uint32(b[0]) | uint32(b[1])<<8 | uint32(b[2])<<16 | uint32(b[3])<<24
h *= m
h ^= h >> 16
}
switch len(b) {
case 3:
h += uint32(b[2]) << 16
fallthrough
case 2:
h += uint32(b[1]) << 8
fallthrough
case 1:
h += uint32(b[0])
h *= m
h ^= h >> 24
}
return h
}
// FilterPolicy implements the db.FilterPolicy interface from the leveldb/db
// package.
//
// The integer value is the approximate number of bits used per key. A good
// value is 10, which yields a filter with ~ 1% false positive rate.
//
// It is valid to use the other API in this package (leveldb/bloom) without
// using this type or the leveldb/db package.
// type FilterPolicy int
// // Name implements the db.FilterPolicy interface.
// func (p FilterPolicy) Name() string {
// // This string looks arbitrary, but its value is written to LevelDB .ldb
// // files, and should be this exact value to be compatible with those files
// // and with the C++ LevelDB code.
// return "leveldb.BuiltinBloomFilter2"
// }
// // AppendFilter implements the db.FilterPolicy interface.
// func (p FilterPolicy) AppendFilter(dst []byte, keys [][]byte) []byte {
// return appendFilter(dst, keys, int(p))
// }
// // MayContain implements the db.FilterPolicy interface.
// func (p FilterPolicy) MayContain(filter, key []byte) bool {
// return Filter(filter).MayContain(key)
// }
y/bloom_test.go
+147
View File
@@ -0,0 +1,147 @@
// Copyright 2013 The LevelDB-Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package y
import (
"testing"
)
func (f Filter) String() string {
s := make([]byte, 8*len(f))
for i, x := range f {
for j := 0; j < 8; j++ {
if x&(1<<uint(j)) != 0 {
s[8*i+j] = '1'
} else {
s[8*i+j] = '.'
}
}
}
return string(s)
}
func TestSmallBloomFilter(t *testing.T) {
var hash []uint32
for _, word := range [][]byte{
[]byte("hello"),
[]byte("world"),
} {
hash = append(hash, Hash(word))
}
f := NewFilter(hash, 10)
got := f.String()
// The magic want string comes from running the C++ leveldb code's bloom_test.cc.
want := "1...1.........1.........1.....1...1...1.....1.........1.....1....11....."
if got != want {
t.Fatalf("bits:\ngot %q\nwant %q", got, want)
}
m := map[string]bool{
"hello": true,
"world": true,
"x": false,
"foo": false,
}
for k, want := range m {
got := f.MayContainKey([]byte(k))
if got != want {
t.Errorf("MayContain: k=%q: got %v, want %v", k, got, want)
}
}
}
func TestBloomFilter(t *testing.T) {
nextLength := func(x int) int {
if x < 10 {
return x + 1
}
if x < 100 {
return x + 10
}
if x < 1000 {
return x + 100
}
return x + 1000
}
le32 := func(i int) []byte {
b := make([]byte, 4)
b[0] = uint8(uint32(i) >> 0)
b[1] = uint8(uint32(i) >> 8)
b[2] = uint8(uint32(i) >> 16)
b[3] = uint8(uint32(i) >> 24)
return b
}
nMediocreFilters, nGoodFilters := 0, 0
loop:
for length := 1; length <= 10000; length = nextLength(length) {
keys := make([][]byte, 0, length)
for i := 0; i < length; i++ {
keys = append(keys, le32(i))
}
var hashes []uint32
for _, key := range keys {
hashes = append(hashes, Hash(key))
}
f := NewFilter(hashes, 10)
if len(f) > (length*10/8)+40 {
t.Errorf("length=%d: len(f)=%d is too large", length, len(f))
continue
}
// All added keys must match.
for _, key := range keys {
if !f.MayContainKey(key) {
t.Errorf("length=%d: did not contain key %q", length, key)
continue loop
}
}
// Check false positive rate.
nFalsePositive := 0
for i := 0; i < 10000; i++ {
if f.MayContainKey(le32(1e9 + i)) {
nFalsePositive++
}
}
if nFalsePositive > 0.02*10000 {
t.Errorf("length=%d: %d false positives in 10000", length, nFalsePositive)
continue
}
if nFalsePositive > 0.0125*10000 {
nMediocreFilters++
} else {
nGoodFilters++
}
}
if nMediocreFilters > nGoodFilters/5 {
t.Errorf("%d mediocre filters but only %d good filters", nMediocreFilters, nGoodFilters)
}
}
func TestHash(t *testing.T) {
// The magic want numbers come from running the C++ leveldb code in hash.cc.
testCases := []struct {
s string
want uint32
}{
{"", 0xbc9f1d34},
{"g", 0xd04a8bda},
{"go", 0x3e0b0745},
{"gop", 0x0c326610},
{"goph", 0x8c9d6390},
{"gophe", 0x9bfd4b0a},
{"gopher", 0xa78edc7c},
{"I had a dream it would end this way.", 0xe14a9db9},
}
for _, tc := range testCases {
if got := Hash([]byte(tc.s)); got != tc.want {
t.Errorf("s=%q: got 0x%08x, want 0x%08x", tc.s, got, tc.want)
}
}
}