this repo has no description
Add reverse scan to BTree.Reader
+96
-11
+1
-1
lib/btree/fuzz/model_fuzz.ex
+1
-1
lib/btree/fuzz/model_fuzz.ex
···
150
150
151
151
read_version = Enum.random(max(version - 5000, 0)..version)
152
152
{start_key, end_key} = make_range(opt_key_bits)
153
-
reverse? = false
153
+
reverse? = Enum.random([true, false])
154
154
limit = random_limit()
155
155
#dbg SimpleKV.dump(simple_kv, read_version), limit: :infinity
156
156
#dbg {start_key, end_key, reverse?, limit}
+95
-10
lib/btree/reader.ex
+95
-10
lib/btree/reader.ex
···
45
45
@spec scan(BTree.t, non_neg_integer, binary, binary, boolean, non_neg_integer) :: [{binary, binary}]
46
46
def scan(%BTree{} = btree, version, start_key, end_key, reverse?, limit) do
47
47
assert limit > 0
48
-
merge_it = new_merge_iterator(btree, start_key, reverse?)
49
48
50
49
case reverse? do
51
-
false -> start_scan_forward(merge_it, version, end_key, limit)
50
+
true ->
51
+
merge_it = new_merge_iterator(btree, end_key, reverse?)
52
+
start_scan_backward(merge_it, version, start_key, limit)
53
+
false ->
54
+
merge_it = new_merge_iterator(btree, start_key, reverse?)
55
+
start_scan_forward(merge_it, version, end_key, limit)
52
56
end
53
57
|> Enum.reverse()
54
58
end
55
59
60
+
# Abandon hope all ye who enter here
61
+
defp start_scan_backward(merge_it, version, start_key, limit) do
62
+
{_, _, {[key | ver], val}} = merge_it
63
+
merge_it = merge_backward(merge_it)
64
+
65
+
cond do
66
+
key < start_key ->
67
+
# Reached the end of the requested range without ever finding a visible pair
68
+
[]
69
+
70
+
ver > version ->
71
+
# This pair is not visible because its version is too new, skip it
72
+
start_scan_backward(merge_it, version, start_key, limit)
73
+
74
+
true ->
75
+
# This pair is visible, start scanning with it
76
+
do_scan_backward(merge_it, version, start_key, limit, key, ver, val, [], 0)
77
+
end
78
+
end
79
+
80
+
defp finish_scan_backward(_prev_key, :tombstone, acc), do: acc
81
+
defp finish_scan_backward(prev_key, prev_value, acc), do: [{prev_key, prev_value} | acc]
82
+
83
+
defp do_scan_backward(merge_it, version, start_key, limit, prev_key, prev_ver, prev_val, acc, count) do
84
+
{_, _, {[key | ver], val}} = merge_it
85
+
merge_it = merge_backward(merge_it)
86
+
87
+
cond do
88
+
key < start_key ->
89
+
# Reached the end of the requested range
90
+
finish_scan_backward(prev_key, prev_val, acc)
91
+
92
+
ver > version ->
93
+
# This pair is not visible because its version is too new, skip it
94
+
do_scan_backward(merge_it, version, start_key, limit, prev_key, prev_ver, prev_val, acc, count)
95
+
96
+
key == prev_key ->
97
+
# This is an older version of the same key, skip it and keep the original
98
+
do_scan_backward(merge_it, version, start_key, limit, prev_key, prev_ver, prev_val, acc, count)
99
+
100
+
prev_val == :tombstone ->
101
+
# This is a new key but the previous value was a tombstone, overwrite
102
+
do_scan_backward(merge_it, version, start_key, limit, key, ver, val, acc, count)
103
+
104
+
true ->
105
+
# This pair has a new key, so it is now safe to accumulate the previous key
106
+
# and then check the limit
107
+
#
108
+
# Note: *both* sides of this branch accumulate `prev_key`/`prev_val`
109
+
count = count + 1
110
+
case count < limit do
111
+
true ->
112
+
# We are under the limit, accumulate and keep going
113
+
acc = [{prev_key, prev_val} | acc]
114
+
do_scan_backward(merge_it, version, start_key, limit, key, ver, val, acc, count)
115
+
false ->
116
+
# We have reached the limit, accumulate the last pair and finish
117
+
finish_scan_backward(prev_key, prev_val, acc)
118
+
end
119
+
end
120
+
end
121
+
56
122
defp start_scan_forward(:empty = _merge_it, _version, _end_key, _limit), do: []
57
123
defp start_scan_forward(merge_it, version, end_key, limit) do
58
124
{_, _, {[key | ver], val}} = merge_it
···
94
160
do_scan_forward(merge_it, version, end_key, limit, prev_key, prev_ver, prev_val, acc, count)
95
161
96
162
key == prev_key ->
97
-
# This is a newer (but visible) version of the same key, overwrite
163
+
# This is a newer (but visible) version of the previous key, overwrite
98
164
do_scan_forward(merge_it, version, end_key, limit, key, ver, val, acc, count)
99
165
100
166
prev_val == :tombstone ->
···
129
195
merge_it = {storage_it, versioned_it, nil}
130
196
131
197
case reverse? do
198
+
true -> merge_backward(merge_it)
132
199
false -> merge_forward(merge_it)
133
200
end
134
201
end
135
202
203
+
defp merge_backward({storage_it, versioned_it, _current}) do
204
+
{[storage_key | _ver], _value} = storage_pair = storage_it.current_pair
205
+
{_vt, _rv?, {[versioned_key | _ver], _value} = versioned_pair} = versioned_it
206
+
207
+
cond do
208
+
storage_key == :infinity and versioned_key == :infinity -> :empty
209
+
210
+
versioned_key == :infinity -> {next_backward(storage_it), versioned_it, storage_pair}
211
+
storage_key == :infinity -> {storage_it, versioned_backward(versioned_it), versioned_pair}
212
+
storage_key <= versioned_key -> {storage_it, versioned_backward(versioned_it), versioned_pair}
213
+
true -> {next_backward(storage_it), versioned_it, storage_pair}
214
+
end
215
+
end
216
+
136
217
defp merge_forward({storage_it, versioned_it, _current}) do
137
218
{[storage_key | _ver], _value} = storage_pair = storage_it.current_pair
138
219
{_vt, _rv?, {[versioned_key | _ver], _value} = versioned_pair} = versioned_it
139
220
140
221
cond do
141
-
storage_key == :infinity and versioned_key == :infinity ->
142
-
:empty
222
+
storage_key == :infinity and versioned_key == :infinity -> :empty
143
223
144
224
versioned_key == :infinity -> {next_forward(storage_it), versioned_it, storage_pair}
145
225
storage_key == :infinity -> {storage_it, versioned_forward(versioned_it), versioned_pair}
···
150
230
151
231
defp new_versioned_iterator(versioned_tree, start_key, reverse?) do
152
232
case reverse? do
153
-
true -> raise "todo"
233
+
true -> versioned_backward({versioned_tree, true, {[start_key | :infinity], nil}})
154
234
false -> versioned_forward({versioned_tree, false, {[start_key | -1], nil}})
155
235
end
156
236
end
157
237
238
+
defp versioned_backward({versioned_tree, reverse?, {current_key, _value}}) do
239
+
case :ets.prev_lookup(versioned_tree, current_key) do
240
+
{_key, [pair]} -> {versioned_tree, reverse?, pair}
241
+
:"$end_of_table" -> {versioned_tree, reverse?, {[:infinity | 0], :infinity}}
242
+
end
243
+
end
244
+
158
245
defp versioned_forward({versioned_tree, reverse?, {current_key, _value}}) do
159
246
case :ets.next_lookup(versioned_tree, current_key) do
160
-
{_key, [pair]} ->
161
-
{versioned_tree, reverse?, pair}
162
-
:"$end_of_table" ->
163
-
{versioned_tree, reverse?, {[:infinity | 0], :infinity}}
247
+
{_key, [pair]} -> {versioned_tree, reverse?, pair}
248
+
:"$end_of_table" -> {versioned_tree, reverse?, {[:infinity | 0], :infinity}}
164
249
end
165
250
end
166
251