herald.go/db/iteroptions.go
Jonathan Moody 071aa2a7ad
Catchup to python-herald schema. Plus lots of refactoring. (#49)
* 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.
2022-08-16 08:45:41 +03:00

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
}