/
reader.go
95 lines (86 loc) · 2.03 KB
/
reader.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
// Copyright 2015 Reborndb Org. All Rights Reserved.
// Licensed under the MIT (MIT-LICENSE.txt) license.
package store
import (
"bytes"
"github.com/reborndb/go/redis/rdb"
"github.com/reborndb/qdb/pkg/engine"
)
type storeIterator struct {
engine.Iterator
serial uint64
}
type storeReader interface {
getRowValue(key []byte) ([]byte, error)
getIterator() *storeIterator
putIterator(it *storeIterator)
}
func loadObjEntry(r storeReader, db uint32, key []byte) (storeRow, *rdb.ObjEntry, error) {
o, err := loadStoreRow(r, db, key)
if err != nil || o == nil {
return o, nil, err
}
if o.IsExpired() {
return o, nil, nil
}
if val, err := o.loadObjectValue(r); err != nil {
return o, nil, err
} else {
obj := &rdb.ObjEntry{
DB: db,
Key: key,
Value: val,
ExpireAt: uint64(o.GetExpireAt()),
}
return o, obj, nil
}
}
func loadBinEntry(r storeReader, db uint32, key []byte) (storeRow, *rdb.BinEntry, error) {
o, obj, err := loadObjEntry(r, db, key)
if err != nil || obj == nil {
return o, nil, err
}
if bin, err := obj.BinEntry(); err != nil {
return o, nil, err
} else {
return o, bin, nil
}
}
func firstKeyUnderSlot(r storeReader, db uint32, slot uint32) ([]byte, error) {
it := r.getIterator()
defer r.putIterator(it)
pfx := EncodeMetaKeyPrefixSlot(db, slot)
if it.SeekTo(pfx); it.Valid() {
metaKey := it.Key()
if !bytes.HasPrefix(metaKey, pfx) {
return nil, it.Error()
}
_, key, err := DecodeMetaKey(metaKey)
if err != nil {
return nil, err
}
return key, it.Error()
}
return nil, it.Error()
}
func allKeysWithTag(r storeReader, db uint32, tag []byte) ([][]byte, error) {
it := r.getIterator()
defer r.putIterator(it)
var keys [][]byte
pfx := EncodeMetaKeyPrefixTag(db, tag)
for it.SeekTo(pfx); it.Valid(); it.Next() {
metaKey := it.Key()
if !bytes.HasPrefix(metaKey, pfx) {
break
}
_, key, err := DecodeMetaKey(metaKey)
if err != nil {
return nil, err
}
keys = append(keys, key)
}
if err := it.Error(); err != nil {
return nil, err
}
return keys, nil
}