-
Notifications
You must be signed in to change notification settings - Fork 1.2k
/
table_search.go
223 lines (186 loc) · 4.2 KB
/
table_search.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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
package mergeset
import (
"bytes"
"container/heap"
"fmt"
"io"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
)
// TableSearch is a reusable cursor used for searching in the Table.
type TableSearch struct {
// Item contains the next item after successful NextItem
// or FirstItemWithPrefix call.
//
// Item contents breaks after the next call to NextItem.
Item []byte
tb *Table
pws []*partWrapper
psPool []partSearch
psHeap partSearchHeap
err error
nextItemNoop bool
needClosing bool
}
func (ts *TableSearch) reset() {
ts.Item = nil
ts.tb = nil
for i := range ts.pws {
ts.pws[i] = nil
}
ts.pws = ts.pws[:0]
for i := range ts.psPool {
ts.psPool[i].reset()
}
ts.psPool = ts.psPool[:0]
for i := range ts.psHeap {
ts.psHeap[i] = nil
}
ts.psHeap = ts.psHeap[:0]
ts.err = nil
ts.nextItemNoop = false
ts.needClosing = false
}
// Init initializes ts for searching in the tb.
//
// MustClose must be called when the ts is no longer needed.
func (ts *TableSearch) Init(tb *Table, shouldCacheBlock func(item []byte) bool) {
if ts.needClosing {
logger.Panicf("BUG: missing MustClose call before the next call to Init")
}
ts.reset()
ts.tb = tb
ts.needClosing = true
ts.pws = ts.tb.getParts(ts.pws[:0])
// Initialize the psPool.
if n := len(ts.pws) - cap(ts.psPool); n > 0 {
ts.psPool = append(ts.psPool[:cap(ts.psPool)], make([]partSearch, n)...)
}
ts.psPool = ts.psPool[:len(ts.pws)]
for i, pw := range ts.pws {
ts.psPool[i].Init(pw.p, shouldCacheBlock)
}
}
// Seek seeks for the first item greater or equal to k in the ts.
func (ts *TableSearch) Seek(k []byte) {
if err := ts.Error(); err != nil {
// Do nothing on unrecoverable error.
return
}
ts.err = nil
// Initialize the psHeap.
var errors []error
ts.psHeap = ts.psHeap[:0]
for i := range ts.psPool {
ps := &ts.psPool[i]
ps.Seek(k)
if !ps.NextItem() {
if err := ps.Error(); err != nil {
errors = append(errors, err)
}
continue
}
ts.psHeap = append(ts.psHeap, ps)
}
if len(errors) > 0 {
// Return only the first error, since it has no sense in returning all errors.
ts.err = fmt.Errorf("cannot seek %q: %w", k, errors[0])
return
}
if len(ts.psHeap) == 0 {
ts.err = io.EOF
return
}
heap.Init(&ts.psHeap)
ts.Item = ts.psHeap[0].Item
ts.nextItemNoop = true
}
// FirstItemWithPrefix seeks for the first item with the given prefix in the ts.
//
// It returns io.EOF if such an item doesn't exist.
func (ts *TableSearch) FirstItemWithPrefix(prefix []byte) error {
ts.Seek(prefix)
if !ts.NextItem() {
if err := ts.Error(); err != nil {
return err
}
return io.EOF
}
if err := ts.Error(); err != nil {
return err
}
if !bytes.HasPrefix(ts.Item, prefix) {
return io.EOF
}
return nil
}
// NextItem advances to the next item.
func (ts *TableSearch) NextItem() bool {
if ts.err != nil {
return false
}
if ts.nextItemNoop {
ts.nextItemNoop = false
return true
}
ts.err = ts.nextBlock()
if ts.err != nil {
if ts.err != io.EOF {
ts.err = fmt.Errorf("cannot obtain the next block to search in the table: %w", ts.err)
}
return false
}
return true
}
func (ts *TableSearch) nextBlock() error {
psMin := ts.psHeap[0]
if psMin.NextItem() {
heap.Fix(&ts.psHeap, 0)
ts.Item = ts.psHeap[0].Item
return nil
}
if err := psMin.Error(); err != nil {
return err
}
heap.Pop(&ts.psHeap)
if len(ts.psHeap) == 0 {
return io.EOF
}
ts.Item = ts.psHeap[0].Item
return nil
}
// Error returns the last error in ts.
func (ts *TableSearch) Error() error {
if ts.err == io.EOF {
return nil
}
return ts.err
}
// MustClose closes the ts.
func (ts *TableSearch) MustClose() {
if !ts.needClosing {
logger.Panicf("BUG: missing Init call before MustClose call")
}
ts.tb.putParts(ts.pws)
ts.reset()
}
type partSearchHeap []*partSearch
func (psh *partSearchHeap) Len() int {
return len(*psh)
}
func (psh *partSearchHeap) Less(i, j int) bool {
x := *psh
return string(x[i].Item) < string(x[j].Item)
}
func (psh *partSearchHeap) Swap(i, j int) {
x := *psh
x[i], x[j] = x[j], x[i]
}
func (psh *partSearchHeap) Push(x interface{}) {
*psh = append(*psh, x.(*partSearch))
}
func (psh *partSearchHeap) Pop() interface{} {
a := *psh
v := a[len(a)-1]
*psh = a[:len(a)-1]
return v
}