LBRYCommunity/herald.go
1
0
Fork 0
Code Issues 16 Pull requests Projects Releases 14 Packages Wiki Activity

Add pluggable SerializationAPI. Use it in prefixes_test.

Browse source 
Populate PrefixRowKV.RawKey,RawValue when appropriate.
This commit is contained in:
Jonathan Moody 2022-08-10 10:52:05 -04:00
parent 441da6b2b3
commit 02f7a41d49
5 changed files with 140 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

BIN
db/.DS_Store vendored Normal file
View file

Binary file not shown.

12
db/db.go
View file

@ -850,8 +850,8 @@ func ReadPrefixN(db *grocksdb.DB, prefix []byte, n int) []*prefixes.PrefixRowKV
value := it.Value()
res[i] = &prefixes.PrefixRowKV{
Key: key.Data(),
Value: value.Data(),
RawKey: key.Data(),
RawValue: value.Data(),
}
key.Free()
@ -908,8 +908,8 @@ func readWriteRawNCF(db *grocksdb.DB, options *IterOptions, out string, n int, f
if i >= n {
return
}
key := kv.Key.([]byte)
value := kv.Value.([]byte)
key := kv.RawKey
value := kv.RawValue
keyHex := hex.EncodeToString(key)
valueHex := hex.EncodeToString(value)
//log.Println(keyHex)
@ -947,8 +947,8 @@ func ReadWriteRawN(db *grocksdb.DB, options *IterOptions, out string, n int) {
if i >= n {
return
}
key := kv.Key.([]byte)
value := kv.Value.([]byte)
key := kv.RawKey
value := kv.RawValue
keyHex := hex.EncodeToString(key)
valueHex := hex.EncodeToString(value)
log.Println(keyHex)

27
db/iteroptions.go
View file

@ -24,6 +24,7 @@ type IterOptions struct {
RawValue bool
CfHandle *grocksdb.ColumnFamilyHandle
It *grocksdb.Iterator
Serializer *prefixes.SerializationAPI
}
// NewIterateOptions creates a defualt options structure for a db iterator.
@ -41,6 +42,7 @@ func NewIterateOptions() *IterOptions {
RawValue: false,
CfHandle: nil,
It: nil,
Serializer: prefixes.ProductionAPI,
}
}
@ -99,6 +101,11 @@ func (o *IterOptions) WithRawValue(rawValue bool) *IterOptions {
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
@ -117,8 +124,10 @@ func (opts *IterOptions) ReadRow(prevKey *[]byte) *prefixes.PrefixRowKV {
valueData := value.Data()
valueLen := len(valueData)
var outKey interface{} = nil
var outValue interface{} = nil
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)
@ -136,12 +145,12 @@ func (opts *IterOptions) ReadRow(prevKey *[]byte) *prefixes.PrefixRowKV {
newKeyData := make([]byte, keyLen)
copy(newKeyData, keyData)
if opts.IncludeKey && !opts.RawKey {
outKey, err = prefixes.UnpackGenericKey(newKeyData)
outKey, err = opts.Serializer.UnpackKey(newKeyData)
if err != nil {
log.Error(err)
}
} else if opts.IncludeKey {
outKey = newKeyData
rawOutKey = newKeyData
}
// Value could be quite large, so this setting could be important
@ -150,18 +159,20 @@ func (opts *IterOptions) ReadRow(prevKey *[]byte) *prefixes.PrefixRowKV {
newValueData := make([]byte, valueLen)
copy(newValueData, valueData)
if !opts.RawValue {
outValue, err = prefixes.UnpackGenericValue(newKeyData, newValueData)
outValue, err = opts.Serializer.UnpackValue(newKeyData, newValueData)
if err != nil {
log.Error(err)
}
} else {
outValue = newValueData
rawOutValue = newValueData
}
}
kv := &prefixes.PrefixRowKV{
Key: outKey,
Value: outValue,
Key: outKey,
Value: outValue,
RawKey: rawOutKey,
RawValue: rawOutValue,
}
*prevKey = newKeyData

80
db/prefixes/generic.go
View file

@ -18,6 +18,7 @@ type tableMeta struct {
newValue func() interface{}
newKeyUnpack func([]byte) interface{}
newValueUnpack func([]byte) interface{}
API *SerializationAPI
}
var tableRegistry = map[byte]tableMeta{
@ -556,3 +557,82 @@ func GenericUnpack(pfx []byte, key bool, buf []byte) (interface{}, error) {
}
return kv, nil
}
func GetSerializationAPI(prefix []byte) *SerializationAPI {
t, ok := tableRegistry[prefix[0]]
if !ok {
panic(fmt.Sprintf("not handled: prefix=%v", prefix))
}
if t.API != nil {
return t.API
}
return ProductionAPI
}
type SerializationAPI struct {
PackKey func(key BaseKey) ([]byte, error)
PackPartialKey func(key BaseKey, fields int) ([]byte, error)
PackValue func(value BaseValue) ([]byte, error)
UnpackKey func(key []byte) (BaseKey, error)
UnpackValue func(prefix []byte, value []byte) (BaseValue, error)
}
var ProductionAPI = &SerializationAPI{
PackKey: PackGenericKey,
PackPartialKey: PackPartialGenericKey,
PackValue: PackGenericValue,
UnpackKey: UnpackGenericKey,
UnpackValue: UnpackGenericValue,
}
var RegressionAPI_1 = &SerializationAPI{
PackKey: func(key BaseKey) ([]byte, error) {
return GenericPack(key, -1)
},
PackPartialKey: func(key BaseKey, fields int) ([]byte, error) {
return GenericPack(key, fields)
},
PackValue: func(value BaseValue) ([]byte, error) {
return GenericPack(value, -1)
},
UnpackKey: func(key []byte) (BaseKey, error) {
return UnpackGenericKey(key)
},
UnpackValue: func(prefix []byte, value []byte) (BaseValue, error) {
return UnpackGenericValue(prefix, value)
},
}
var RegressionAPI_2 = &SerializationAPI{
PackKey: PackGenericKey,
PackPartialKey: PackPartialGenericKey,
PackValue: PackGenericValue,
UnpackKey: func(key []byte) (BaseKey, error) {
k, err := GenericUnpack(key, true, key)
return k.(BaseKey), err
},
UnpackValue: func(prefix []byte, value []byte) (BaseValue, error) {
k, err := GenericUnpack(prefix, false, value)
return k.(BaseValue), err
},
}
var RegressionAPI_3 = &SerializationAPI{
PackKey: func(key BaseKey) ([]byte, error) {
return GenericPack(key, -1)
},
PackPartialKey: func(key BaseKey, fields int) ([]byte, error) {
return GenericPack(key, fields)
},
PackValue: func(value BaseValue) ([]byte, error) {
return GenericPack(value, -1)
},
UnpackKey: func(key []byte) (BaseKey, error) {
k, err := GenericUnpack(key, true, key)
return k.(BaseKey), err
},
UnpackValue: func(prefix []byte, value []byte) (BaseValue, error) {
k, err := GenericUnpack(prefix, false, value)
return k.(BaseValue), err
},
}

41
db/prefixes/prefixes_test.go
View file

@ -61,6 +61,23 @@ func testInit(filePath string) (*grocksdb.DB, [][]string, func(), *grocksdb.Colu
}
func testGeneric(filePath string, prefix byte, numPartials int) func(*testing.T) {
return func(t *testing.T) {
APIs := []*prefixes.SerializationAPI{
prefixes.GetSerializationAPI([]byte{prefix}),
// Verify combinations of production vs. "restruct" implementations of
// serialization API (e.g production Pack() with "restruct" Unpack()).
prefixes.RegressionAPI_1,
prefixes.RegressionAPI_2,
prefixes.RegressionAPI_3,
}
for _, api := range APIs {
opts := dbpkg.NewIterateOptions().WithPrefix([]byte{prefix}).WithSerializer(api).WithIncludeValue(true)
testGenericOptions(opts, filePath, prefix, numPartials)(t)
}
}
}
func testGenericOptions(options *dbpkg.IterOptions, filePath string, prefix byte, numPartials int) func(*testing.T) {
return func(t *testing.T) {
wOpts := grocksdb.NewDefaultWriteOptions()
@ -79,26 +96,26 @@ func testGeneric(filePath string, prefix byte, numPartials int) func(*testing.T)
db.PutCF(wOpts, handle, key, val)
}
// test prefix
options := dbpkg.NewIterateOptions().WithPrefix([]byte{prefix}).WithIncludeValue(true)
options = options.WithCfHandle(handle)
ch := dbpkg.IterCF(db, options)
var i = 0
for kv := range ch {
// log.Println(kv.Key)
gotKey, err := prefixes.PackGenericKey(prefix, kv.Key)
gotKey, err := options.Serializer.PackKey(kv.Key)
if err != nil {
log.Println(err)
}
for j := 1; j <= numPartials; j++ {
keyPartial, _ := prefixes.PackPartialGenericKey(prefix, kv.Key, j)
keyPartial, _ := options.Serializer.PackPartialKey(kv.Key, j)
// Check pack partial for sanity
if !bytes.HasPrefix(gotKey, keyPartial) {
// || (!bytes.HasSuffix(gotKey, []byte{0}) && bytes.Equal(gotKey, keyPartial))
t.Errorf("%+v should be prefix of %+v\n", keyPartial, gotKey)
}
}
got, err := prefixes.PackGenericValue(prefix, kv.Value)
got, err := options.Serializer.PackValue(kv.Value)
if err != nil {
log.Println(err)
}
@ -133,12 +150,12 @@ func testGeneric(filePath string, prefix byte, numPartials int) func(*testing.T)
if err != nil {
log.Println(err)
}
options2 := dbpkg.NewIterateOptions().WithStart(start).WithStop(stop).WithIncludeValue(true)
options2 := dbpkg.NewIterateOptions().WithSerializer(options.Serializer).WithStart(start).WithStop(stop).WithIncludeValue(true)
options2 = options2.WithCfHandle(handle)
ch2 := dbpkg.IterCF(db, options2)
i = 0
for kv := range ch2 {
got, err := prefixes.PackGenericValue(prefix, kv.Value)
got, err := options2.Serializer.PackValue(kv.Value)
if err != nil {
log.Println(err)
}
@ -340,3 +357,15 @@ func TestUTXOKey_String(t *testing.T) {
})
}
}
func TestHashXStatus(t *testing.T) {
filePath := fmt.Sprintf("../../testdata/%c.csv", prefixes.HashXStatus)
key := &prefixes.HashXStatusKey{}
testGeneric(filePath, prefixes.HashXStatus, key.NumFields())(t)
}
func TestHashXMempoolStatus(t *testing.T) {
filePath := fmt.Sprintf("../../testdata/%c.csv", prefixes.HashXMempoolStatus)
key := &prefixes.HashXMempoolStatusKey{}
testGeneric(filePath, prefixes.HashXMempoolStatus, key.NumFields())(t)
}