071aa2a7ad
* Make prefixes_test.go more resilient against garbage left by a prior crash. Also correct error logging. * Don't do the ones' complement thing with DBStateValue fields HistFlushCount, CompFlushCount, CompCursor. Python-herald doesn't do it, and it presents one more irregular case for (un)marshalling fields. * Simplify type-specific partial packing, and simplify dispatch for pack key/value. * Add struct field annotations and refactor to prepare for use of "restruct" generic packing/unpacking. * Add dynamic pack/unpack based on "restruct" module. Dispatch normal pack/unpack through tableRegistry[] map instead of switch. * Add 5 new prefixes/tables (TrendingNotifications..HashXMempoolStatus). * Undo rename. TouchedOrDeleted -> ClaimDiff. * Fixup callers of eliminated partial pack functions. Have them use key.PartialPack(n). * Add pluggable SerializationAPI. Use it in prefixes_test. Populate PrefixRowKV.RawKey,RawValue when appropriate. * Undo accidental bump of rocksdb version. * Add .vscode dir to gitignore. * Fix ClaimToChannelValue annotation. Implement BlockTxsValue workaround as I can't find the right annotation to get it marshalled/unmarshalled. * Strengthen partial packing verification. Fix bugs in UnpackKey/UnpackValue for new types. * Remove .DS_Store, and ignore in future. * Fix MempoolTxKey, TouchedHashXValue. Remove some unneeded struct tags. * Generate test data and complete the tests for the new tables. Add Fuzz tests for TouchedHashXKey, TouchedHashXValue with happy path test data (only). * Move tableRegistry to prefixes.go and rename it prefixRegistry. Other minor fixes, comments. * Add test that runs through GetPrefixes() contents, and verifies they are registered in prefixRegistry.
201 lines
4.6 KiB
Go
201 lines
4.6 KiB
Go
package db
|
|
|
|
// iteroptions.go contains the implementation for iterators on rocksdb used by the hub
|
|
|
|
import (
|
|
"bytes"
|
|
|
|
"github.com/lbryio/herald.go/db/prefixes"
|
|
"github.com/linxGnu/grocksdb"
|
|
|
|
log "github.com/sirupsen/logrus"
|
|
)
|
|
|
|
type IterOptions struct {
|
|
FillCache bool
|
|
Prefix []byte
|
|
Start []byte //interface{}
|
|
Stop []byte //interface{}
|
|
IncludeStart bool
|
|
IncludeStop bool
|
|
IncludeKey bool
|
|
IncludeValue bool
|
|
RawKey bool
|
|
RawValue bool
|
|
CfHandle *grocksdb.ColumnFamilyHandle
|
|
It *grocksdb.Iterator
|
|
Serializer *prefixes.SerializationAPI
|
|
}
|
|
|
|
// NewIterateOptions creates a defualt options structure for a db iterator.
|
|
func NewIterateOptions() *IterOptions {
|
|
return &IterOptions{
|
|
FillCache: false,
|
|
Prefix: []byte{},
|
|
Start: nil,
|
|
Stop: nil,
|
|
IncludeStart: true,
|
|
IncludeStop: false,
|
|
IncludeKey: true,
|
|
IncludeValue: false,
|
|
RawKey: false,
|
|
RawValue: false,
|
|
CfHandle: nil,
|
|
It: nil,
|
|
Serializer: prefixes.ProductionAPI,
|
|
}
|
|
}
|
|
|
|
func (o *IterOptions) WithCfHandle(cfHandle *grocksdb.ColumnFamilyHandle) *IterOptions {
|
|
o.CfHandle = cfHandle
|
|
return o
|
|
}
|
|
|
|
func (o *IterOptions) WithFillCache(fillCache bool) *IterOptions {
|
|
o.FillCache = fillCache
|
|
return o
|
|
}
|
|
|
|
func (o *IterOptions) WithPrefix(prefix []byte) *IterOptions {
|
|
o.Prefix = prefix
|
|
return o
|
|
}
|
|
|
|
func (o *IterOptions) WithStart(start []byte) *IterOptions {
|
|
o.Start = start
|
|
return o
|
|
}
|
|
|
|
func (o *IterOptions) WithStop(stop []byte) *IterOptions {
|
|
o.Stop = stop
|
|
return o
|
|
}
|
|
|
|
func (o *IterOptions) WithIncludeStart(includeStart bool) *IterOptions {
|
|
o.IncludeStart = includeStart
|
|
return o
|
|
}
|
|
|
|
func (o *IterOptions) WithIncludeStop(includeStop bool) *IterOptions {
|
|
o.IncludeStop = includeStop
|
|
return o
|
|
}
|
|
|
|
func (o *IterOptions) WithIncludeKey(includeKey bool) *IterOptions {
|
|
o.IncludeKey = includeKey
|
|
return o
|
|
}
|
|
|
|
func (o *IterOptions) WithIncludeValue(includeValue bool) *IterOptions {
|
|
o.IncludeValue = includeValue
|
|
return o
|
|
}
|
|
|
|
func (o *IterOptions) WithRawKey(rawKey bool) *IterOptions {
|
|
o.RawKey = rawKey
|
|
return o
|
|
}
|
|
|
|
func (o *IterOptions) WithRawValue(rawValue bool) *IterOptions {
|
|
o.RawValue = rawValue
|
|
return o
|
|
}
|
|
|
|
func (o *IterOptions) WithSerializer(serializer *prefixes.SerializationAPI) *IterOptions {
|
|
o.Serializer = serializer
|
|
return o
|
|
}
|
|
|
|
// ReadRow reads a row from the db, returns nil when no more rows are available.
|
|
func (opts *IterOptions) ReadRow(prevKey *[]byte) *prefixes.PrefixRowKV {
|
|
it := opts.It
|
|
if !it.Valid() {
|
|
log.Trace("ReadRow iterator not valid returning nil")
|
|
return nil
|
|
}
|
|
|
|
key := it.Key()
|
|
defer key.Free()
|
|
keyData := key.Data()
|
|
keyLen := len(keyData)
|
|
|
|
value := it.Value()
|
|
defer value.Free()
|
|
valueData := value.Data()
|
|
valueLen := len(valueData)
|
|
|
|
var outKey prefixes.BaseKey = nil
|
|
var outValue prefixes.BaseValue = nil
|
|
var rawOutKey []byte = nil
|
|
var rawOutValue []byte = nil
|
|
var err error = nil
|
|
|
|
log.Trace("keyData:", keyData)
|
|
log.Trace("valueData:", valueData)
|
|
|
|
// We need to check the current key if we're not including the stop
|
|
// key.
|
|
if !opts.IncludeStop && opts.StopIteration(keyData) {
|
|
log.Trace("ReadRow returning nil")
|
|
return nil
|
|
}
|
|
|
|
// We have to copy the key no matter what because we need to check
|
|
// it on the next iterations to see if we're going to stop.
|
|
newKeyData := make([]byte, keyLen)
|
|
copy(newKeyData, keyData)
|
|
if opts.IncludeKey && !opts.RawKey {
|
|
outKey, err = opts.Serializer.UnpackKey(newKeyData)
|
|
if err != nil {
|
|
log.Error(err)
|
|
}
|
|
} else if opts.IncludeKey {
|
|
rawOutKey = newKeyData
|
|
}
|
|
|
|
// Value could be quite large, so this setting could be important
|
|
// for performance in some cases.
|
|
if opts.IncludeValue {
|
|
newValueData := make([]byte, valueLen)
|
|
copy(newValueData, valueData)
|
|
if !opts.RawValue {
|
|
outValue, err = opts.Serializer.UnpackValue(newKeyData, newValueData)
|
|
if err != nil {
|
|
log.Error(err)
|
|
}
|
|
} else {
|
|
rawOutValue = newValueData
|
|
}
|
|
}
|
|
|
|
kv := &prefixes.PrefixRowKV{
|
|
Key: outKey,
|
|
Value: outValue,
|
|
RawKey: rawOutKey,
|
|
RawValue: rawOutValue,
|
|
}
|
|
*prevKey = newKeyData
|
|
|
|
return kv
|
|
}
|
|
|
|
// StopIteration returns true if we've hit the criteria to end iteration on this key
|
|
func (o *IterOptions) StopIteration(key []byte) bool {
|
|
if key == nil {
|
|
return false
|
|
}
|
|
|
|
maxLenStop := intMin(len(key), len(o.Stop))
|
|
maxLenStart := intMin(len(key), len(o.Start))
|
|
if o.Stop != nil &&
|
|
(bytes.HasPrefix(key, o.Stop) || bytes.Compare(o.Stop, key[:maxLenStop]) < 0) {
|
|
return true
|
|
} else if o.Start != nil &&
|
|
bytes.Compare(o.Start, key[:maxLenStart]) > 0 {
|
|
return true
|
|
} else if o.Prefix != nil && !bytes.HasPrefix(key, o.Prefix) {
|
|
return true
|
|
}
|
|
|
|
return false
|
|
}
|