this repo has no description
Refactor HybridKV scans to use range info from other KVs
+160
-121
+73
-76
lib/hybrid_kv.ex
+73
-76
lib/hybrid_kv.ex
···
37
37
@spec scan(%HybridKV{}, non_neg_integer, binary, binary, keyword) :: RangeResult.t
38
38
def scan(%HybridKV{} = kv, version, start_key, end_key, opts \\ []) do
39
39
limit = Keyword.get(opts, :limit, :infinity)
40
+
reverse = Keyword.get(opts, :reverse, false)
40
41
41
42
read_limit = case limit do
42
43
:infinity -> :infinity
43
44
limit -> limit + 1
44
45
end
45
46
46
-
pairs = do_scan_merge(kv, version, start_key, end_key, Keyword.put(opts, :limit, read_limit), 0)
47
+
{pairs, count} =
48
+
case reverse do
49
+
false -> do_scan(:forward, kv, version, start_key, end_key, read_limit, 0)
50
+
true -> do_scan(:backward, kv, version, start_key, end_key, read_limit, 0)
51
+
end
47
52
48
53
# We over-read by 1 and then use the extra read
49
54
# to check if there are more
50
-
# TODO: we could compute count more efficiently in do_scan_merge/do_merge
51
-
count = length(pairs)
52
55
cond do
53
56
count < read_limit ->
54
-
%RangeResult{pairs: pairs, more: false}
57
+
%RangeResult{pairs: pairs, count: count, more: false}
55
58
count == read_limit ->
56
-
%RangeResult{pairs: Enum.take(pairs, limit), more: true}
59
+
%RangeResult{pairs: Enum.take(pairs, limit), count: limit, more: true}
57
60
end
58
61
end
59
62
60
63
# This exists purely as a sanity check and should be unreachable except for bugs
61
-
defp do_scan_merge(_kv, _version, _start_key, _end_key, _opts, 1000), do: raise "Scan caught in loop!"
64
+
defp do_scan(_direction, _kv, _version, _start_key, _end_key, _limit, 1000), do: raise "Scan caught in loop!"
62
65
63
66
# Abandon hope all ye who enter here
64
-
defp do_scan_merge(%HybridKV{} = kv, version, start_key, end_key, opts, scan_count) do
65
-
limit = Keyword.get(opts, :limit, :infinity)
66
-
reverse = Keyword.get(opts, :reverse, false)
67
-
storage_result = FlatKV.scan(kv.storage_kv, start_key, end_key, opts)
67
+
defp do_scan(:forward, %HybridKV{} = kv, version, start_key, end_key, limit, scan_count) do
68
+
storage_result = FlatKV.scan(kv.storage_kv, start_key, end_key, limit: limit)
69
+
{_start, storage_end_key} = storage_result.range
70
+
71
+
# Read mem only up to the end of the range scanned by storage
72
+
mem_result = MemKV.scan(kv.mem_kv, version, start_key, storage_end_key, limit: limit)
73
+
{_start, mem_end_key} = mem_result.range
74
+
75
+
# Both KVs were scanned up to this key
76
+
# Anything past this key was only scanned by storage and must be discarded
77
+
scanned_end_key = min(storage_end_key, mem_end_key)
78
+
79
+
merged_pairs =
80
+
merge(storage_result.pairs, mem_result.pairs, false)
81
+
# TODO: more efficient to do this in merge/3
82
+
|> Enum.take_while(fn {k, _v} -> k < scanned_end_key end)
83
+
84
+
# TODO: compute count in merge/3
85
+
count = length(merged_pairs)
86
+
87
+
cond do
88
+
count > limit ->
89
+
# We got more pairs than we asked for, which can happen if the
90
+
# storage and mem keys are disjoint (which is fairly likely in practice)
91
+
{Enum.take(merged_pairs, limit), limit}
68
92
69
-
mem_result =
70
-
case storage_result.more do
71
-
true ->
72
-
# If we hit the limit in the storage result, scan only the (limited) range it spans (including the last key)
73
-
{last_storage_key, _value} = List.last(storage_result.pairs)
74
-
case reverse do
75
-
false -> MemKV.scan(kv.mem_kv, version, start_key, last_storage_key <> "\x00", opts)
76
-
true -> MemKV.scan(kv.mem_kv, version, last_storage_key, end_key, opts)
77
-
end
78
-
false ->
79
-
# If we *did not* hit the limit, scan the same full range
80
-
MemKV.scan(kv.mem_kv, version, start_key, end_key, opts)
81
-
end
93
+
count == limit ->
94
+
# We got exactly what we wanted!
95
+
{merged_pairs, count}
82
96
83
-
case storage_result.more or mem_result.more do
84
-
false ->
85
-
# We fully scanned [start_key, end_key) for both storage and mem
86
-
merged_pairs = merge(storage_result.pairs, mem_result.pairs, reverse)
87
-
case length(merged_pairs) > limit do
97
+
count < limit ->
98
+
case scanned_end_key == end_key do
88
99
true ->
89
-
# The merged result contains more than `limit` pairs (storage and mem were disjoint)
90
-
Enum.take(merged_pairs, limit)
100
+
# We got <limit pairs but we really did scan the full range
101
+
{merged_pairs, count}
91
102
false ->
92
-
merged_pairs
103
+
# We got <limit pairs and we did not scan the full range, meaning
104
+
# the KVs hit the limit but then the :deleted tombstones cleared
105
+
# out enough keys to bring us back under the limit, so we must
106
+
# keep scanning
107
+
{next_pairs, next_count} = do_scan(:forward, kv, version, scanned_end_key, end_key, limit - count, scan_count + 1)
108
+
{merged_pairs ++ next_pairs, count + next_count}
93
109
end
110
+
end
111
+
end
94
112
95
-
true ->
96
-
# Find the end of the scanned range, i.e. the last key
97
-
# scanned by whichever scan hit the limit, or the
98
-
# "smaller" (taking reverse into account) of the two
99
-
# last keys if *both* scans hit the limit
100
-
scanned_range_end =
101
-
cond do
102
-
storage_result.more and mem_result.more ->
103
-
{sk, _value} = List.last(storage_result.pairs)
104
-
{mk, _value} = List.last(mem_result.pairs)
105
-
case reverse do
106
-
false -> min(sk, mk)
107
-
true -> max(sk, mk)
108
-
end
113
+
# See :forward for comments, the :backward version is the same except key logic
114
+
# is inverted (deals with start_key instead of end_key)
115
+
defp do_scan(:backward, %HybridKV{} = kv, version, start_key, end_key, limit, scan_count) do
116
+
storage_result = FlatKV.scan(kv.storage_kv, start_key, end_key, limit: limit, reverse: true)
117
+
{storage_start_key, _end_key} = storage_result.range
109
118
110
-
storage_result.more ->
111
-
List.last(storage_result.pairs) |> elem(0)
119
+
mem_result = MemKV.scan(kv.mem_kv, version, storage_start_key, end_key, limit: limit, reverse: true)
120
+
{mem_start_key, _end_key} = mem_result.range
112
121
113
-
mem_result.more ->
114
-
List.last(mem_result.pairs) |> elem(0)
115
-
end
122
+
scanned_start_key = max(storage_start_key, mem_start_key)
116
123
117
-
# TODO: it would be more efficient to integrate the scanned_range_end
118
-
# limit into merge/2, but for now this is easier to comprehend
119
-
merged_pairs =
120
-
merge(storage_result.pairs, mem_result.pairs, reverse)
121
-
|> then(fn merged_pairs ->
122
-
case reverse do
123
-
false -> Enum.take_while(merged_pairs, fn {k, _v} -> k <= scanned_range_end end)
124
-
true -> Enum.take_while(merged_pairs, fn {k, _v} -> k >= scanned_range_end end)
125
-
end
126
-
end)
124
+
merged_pairs =
125
+
merge(storage_result.pairs, mem_result.pairs, true)
126
+
|> Enum.take_while(fn {k, _v} -> k >= scanned_start_key end)
127
127
128
-
merged_length = length(merged_pairs)
129
-
cond do
130
-
merged_length > limit ->
131
-
Enum.take(merged_pairs, limit)
128
+
count = length(merged_pairs)
132
129
133
-
merged_length == limit ->
134
-
merged_pairs
130
+
cond do
131
+
count > limit ->
132
+
{Enum.take(merged_pairs, limit), limit}
135
133
136
-
merged_length < limit ->
137
-
# MemKV :deleted tombstones removed enough pairs to bring us under the limit,
138
-
# so we must continue our scan where it left off
139
-
next_opts = Keyword.put(opts, :limit, limit - merged_length)
140
-
next_pairs =
141
-
case reverse do
142
-
false -> do_scan_merge(kv, version, scanned_range_end <> "\x00", end_key, next_opts, scan_count + 1)
143
-
true -> do_scan_merge(kv, version, start_key, scanned_range_end, next_opts, scan_count + 1)
144
-
end
134
+
count == limit ->
135
+
{merged_pairs, count}
145
136
146
-
merged_pairs ++ next_pairs
137
+
count < limit ->
138
+
case scanned_start_key == start_key do
139
+
true ->
140
+
{merged_pairs, count}
141
+
false ->
142
+
{next_pairs, next_count} = do_scan(:backward, kv, version, start_key, scanned_start_key, limit - count, scan_count + 1)
143
+
{merged_pairs ++ next_pairs, count + next_count}
147
144
end
148
145
end
149
146
end
+37
-8
lib/kv/flat_kv.ex
+37
-8
lib/kv/flat_kv.ex
···
47
47
limit = Keyword.get(opts, :limit, :infinity)
48
48
reverse = Keyword.get(opts, :reverse, false)
49
49
50
+
read_limit = case limit do
51
+
:infinity -> :infinity
52
+
limit -> limit + 1
53
+
end
54
+
50
55
# Look up first key
51
56
acc =
52
57
case reverse do
···
59
64
[]
60
65
end
61
66
62
-
result = case reverse do
63
-
false -> do_scan(:forward, kv, end_key, limit, start_key, acc, length(acc))
64
-
true -> do_scan(:backward, kv, start_key, limit, end_key, acc, length(acc))
67
+
{pairs, count} = case reverse do
68
+
false -> do_scan(:forward, kv, end_key, read_limit, start_key, acc, length(acc))
69
+
true -> do_scan(:backward, kv, start_key, read_limit, end_key, acc, length(acc))
70
+
end
71
+
72
+
case reverse do
73
+
false ->
74
+
{pairs, count, scan_end_key, more} =
75
+
cond do
76
+
count < read_limit ->
77
+
{pairs, count, end_key, false}
78
+
count == read_limit ->
79
+
[{last_key, _} | rest] = pairs
80
+
{rest, count - 1, last_key, true}
81
+
end
82
+
%RangeResult{pairs: Enum.reverse(pairs), count: count, range: {start_key, scan_end_key}, more: more}
83
+
84
+
true ->
85
+
{pairs, count, scan_start_key, more} =
86
+
cond do
87
+
count < read_limit ->
88
+
{pairs, count, start_key, false}
89
+
count == read_limit ->
90
+
rest = tl(pairs)
91
+
{last_key, _} = hd(rest)
92
+
{rest, count - 1, last_key, true}
93
+
end
94
+
%RangeResult{pairs: Enum.reverse(pairs), count: count, range: {scan_start_key, end_key}, more: more}
65
95
end
66
-
%RangeResult{result | pairs: Enum.reverse(result.pairs)}
67
96
end
68
97
69
98
defp do_scan(:forward, kv, end_key, limit, prev_key, acc, count) do
···
73
102
false ->
74
103
do_scan(:forward, kv, end_key, limit, key, [{key, value} | acc], count + 1)
75
104
true ->
76
-
%RangeResult{pairs: acc, more: true}
105
+
{acc, count}
77
106
end
78
107
79
108
_ ->
80
-
%RangeResult{pairs: acc, more: false}
109
+
{acc, count}
81
110
end
82
111
end
83
112
···
89
118
false ->
90
119
do_scan(:backward, kv, start_key, limit, key, [{key, value} | acc], count + 1)
91
120
true ->
92
-
%RangeResult{pairs: acc, more: true}
121
+
{acc, count}
93
122
end
94
123
95
124
_ ->
96
-
%RangeResult{pairs: acc, more: false}
125
+
{acc, count}
97
126
end
98
127
end
99
128
+4
-7
lib/kv/test_kv.ex
+4
-7
lib/kv/test_kv.ex
···
70
70
|> Enum.filter(fn {k, _v} -> k >= start_key and k < end_key end)
71
71
|> Enum.sort_by(&elem(&1, 0), direction)
72
72
|> then(fn result ->
73
-
case limit do
74
-
:infinity -> %RangeResult{pairs: result, more: false}
75
-
limit ->
76
-
case length(result) > limit do
77
-
true -> %RangeResult{pairs: Enum.take(result, limit), more: true}
78
-
false -> %RangeResult{pairs: result, more: false}
79
-
end
73
+
count = length(result)
74
+
case count <= limit do
75
+
true -> %RangeResult{pairs: result, count: count, more: false}
76
+
false -> %RangeResult{pairs: Enum.take(result, limit), count: limit, more: true}
80
77
end
81
78
end)
82
79
end
+23
-9
lib/mem_kv.ex
+23
-9
lib/mem_kv.ex
···
66
66
# Note that this means we return all :deleted keys in the range
67
67
# between the last non-deleted key and the key that hit the limit,
68
68
# which is fine (we had to scan them anyway, may as well use them)
69
-
{pairs, more} =
70
-
cond do
71
-
# Covers read_limit == :infinity
72
-
count < read_limit ->
73
-
{pairs, false}
74
-
count == read_limit ->
75
-
{tl(pairs), true}
76
-
end
69
+
case reverse do
70
+
false ->
71
+
{pairs, count, scan_end_key, more} =
72
+
cond do
73
+
count < read_limit ->
74
+
{pairs, count, end_key, false}
75
+
count == read_limit ->
76
+
[{last_key, _} | rest] = pairs
77
+
{rest, count - 1, last_key, true}
78
+
end
79
+
%RangeResult{pairs: Enum.reverse(pairs), count: count, range: {start_key, scan_end_key}, more: more}
77
80
78
-
%RangeResult{pairs: Enum.reverse(pairs), more: more}
81
+
true ->
82
+
{pairs, count, scan_start_key, more} =
83
+
cond do
84
+
count < read_limit ->
85
+
{pairs, count, start_key, false}
86
+
count == read_limit ->
87
+
rest = tl(pairs)
88
+
{last_key, _} = hd(rest)
89
+
{rest, count - 1, last_key, true}
90
+
end
91
+
%RangeResult{pairs: Enum.reverse(pairs), count: count, range: {scan_start_key, end_key}, more: more}
92
+
end
79
93
end
80
94
81
95
defp do_scan(:forward, table, version, end_key, limit, {_, _} = prev, acc, count) do
+3
-1
lib/structs.ex
+3
-1
lib/structs.ex
···
64
64
defmodule RangeResult do
65
65
@type t :: %__MODULE__{
66
66
pairs: [{binary, binary | :deleted | nil}],
67
+
count: non_neg_integer,
68
+
range: {binary, binary} | nil,
67
69
more: boolean,
68
70
}
69
71
@enforce_keys [:pairs, :more]
70
-
defstruct @enforce_keys
72
+
defstruct [:pairs, :count, :range, :more]
71
73
end
72
74
73
75
defmodule ShardStats do
+10
-10
test/kv/flat_kv_test.exs
+10
-10
test/kv/flat_kv_test.exs
···
85
85
{"foo1", "bar1"},
86
86
{"foo2", "bar2"},
87
87
{"foo3", "bar3"},
88
-
], more: false} = FlatKV.scan(kv, "", "foo4")
88
+
], count: 3, range: {"", "foo4"}, more: false} = FlatKV.scan(kv, "", "foo4")
89
89
90
90
assert %RangeResult{pairs: [
91
91
{"foo4", "bar4"},
92
92
{"foo5", "bar5"},
93
93
{"foo6", "bar6"},
94
-
], more: false} = FlatKV.scan(kv, "foo4", "zoo")
94
+
], count: 3, range: {"foo4", "zoo"}, more: false} = FlatKV.scan(kv, "foo4", "zoo")
95
95
96
96
assert %RangeResult{pairs: [
97
97
{"foo2", "bar2"},
98
98
{"foo3", "bar3"},
99
99
{"foo4", "bar4"},
100
100
{"foo5", "bar5"},
101
-
], more: false} = FlatKV.scan(kv, "foo2", "foo6")
101
+
], count: 4, range: {"foo2", "foo6"}, more: false} = FlatKV.scan(kv, "foo2", "foo6")
102
102
103
103
assert %RangeResult{pairs: [
104
104
{"foo1", "bar1"},
···
107
107
{"foo4", "bar4"},
108
108
{"foo5", "bar5"},
109
109
{"foo6", "bar6"},
110
-
], more: false} = FlatKV.scan(kv, "", "zoo")
110
+
], count: 6, range: {"", "zoo"}, more: false} = FlatKV.scan(kv, "", "zoo")
111
111
end
112
112
113
113
test "limits forward", %{kv: kv} do
···
115
115
{"foo1", "bar1"},
116
116
{"foo2", "bar2"},
117
117
{"foo3", "bar3"},
118
-
], more: true} = FlatKV.scan(kv, "", "zoo", limit: 3)
118
+
], count: 3, range: {"", "foo4"}, more: true} = FlatKV.scan(kv, "", "zoo", limit: 3)
119
119
120
120
assert %RangeResult{pairs: [
121
121
{"foo1", "bar1"},
122
122
{"foo2", "bar2"},
123
123
{"foo3", "bar3"},
124
-
], more: false} = FlatKV.scan(kv, "", "foo4", limit: 3)
124
+
], count: 3, range: {"", "foo4"}, more: false} = FlatKV.scan(kv, "", "foo4", limit: 3)
125
125
end
126
126
127
127
test "scans backward", %{kv: kv} do
···
129
129
{"foo3", "bar3"},
130
130
{"foo2", "bar2"},
131
131
{"foo1", "bar1"},
132
-
], more: false} = FlatKV.scan(kv, "", "foo4", reverse: true)
132
+
], count: 3, range: {"", "foo4"}, more: false} = FlatKV.scan(kv, "", "foo4", reverse: true)
133
133
134
134
assert %RangeResult{pairs: [
135
135
{"foo6", "bar6"},
136
136
{"foo5", "bar5"},
137
137
{"foo4", "bar4"},
138
-
], more: false} = FlatKV.scan(kv, "foo4", "zoo", reverse: true)
138
+
], count: 3, range: {"foo4", "zoo"}, more: false} = FlatKV.scan(kv, "foo4", "zoo", reverse: true)
139
139
140
140
assert %RangeResult{pairs: [
141
141
{"foo5", "bar5"},
142
142
{"foo4", "bar4"},
143
143
{"foo3", "bar3"},
144
144
{"foo2", "bar2"},
145
-
], more: false} = FlatKV.scan(kv, "foo2", "foo6", reverse: true)
145
+
], count: 4, range: {"foo2", "foo6"}, more: false} = FlatKV.scan(kv, "foo2", "foo6", reverse: true)
146
146
147
147
assert %RangeResult{pairs: [
148
148
{"foo6", "bar6"},
···
151
151
{"foo3", "bar3"},
152
152
{"foo2", "bar2"},
153
153
{"foo1", "bar1"},
154
-
], more: false} = FlatKV.scan(kv, "", "zoo", reverse: true)
154
+
], count: 6, range: {"", "zoo"}, more: false} = FlatKV.scan(kv, "", "zoo", reverse: true)
155
155
end
156
156
157
157
test "limits backward", %{kv: kv} do
+10
-10
test/mem_kv_test.exs
+10
-10
test/mem_kv_test.exs
···
120
120
{"foo_b", "bar_b"},
121
121
{"foo_c", "bar_c"},
122
122
{"foo_d", "bar_d"},
123
-
], more: false} = MemKV.scan(kv, 1, "foo_a", "foo_z")
123
+
], count: 4, range: {"foo_a", "foo_z"}, more: false} = MemKV.scan(kv, 1, "foo_a", "foo_z")
124
124
125
125
assert %RangeResult{pairs: [
126
126
{"foo_a", "bar_a_2"},
127
127
{"foo_b", "bar_b"},
128
128
{"foo_c", "bar_c"},
129
129
{"foo_d", "bar_d"},
130
-
], more: false} = MemKV.scan(kv, 2, "foo_a", "foo_z")
130
+
], count: 4, range: {"foo_a", "foo_z"}, more: false} = MemKV.scan(kv, 2, "foo_a", "foo_z")
131
131
132
132
assert %RangeResult{pairs: [
133
133
{"foo_a", "bar_a_2"},
···
135
135
{"foo_c", "bar_c"},
136
136
{"foo_d", :deleted},
137
137
{"foo_e", "bar_e_3"},
138
-
], more: false} = MemKV.scan(kv, 3, "foo_a", "foo_z")
138
+
], count: 4, range: {"foo_a", "foo_z"}, more: false} = MemKV.scan(kv, 3, "foo_a", "foo_z")
139
139
end
140
140
141
141
test "limits forward", %{kv: kv} do
···
143
143
{"foo_b", "bar_b_3"},
144
144
{"foo_c", "bar_c"},
145
145
{"foo_d", :deleted},
146
-
], more: true} = MemKV.scan(kv, 3, "foo_b", "foo_z", limit: 2)
146
+
], count: 2, range: {"foo_b", "foo_e"}, more: true} = MemKV.scan(kv, 3, "foo_b", "foo_z", limit: 2)
147
147
148
148
assert %RangeResult{pairs: [
149
149
{"foo_b", "bar_b_3"},
150
150
{"foo_c", "bar_c"},
151
151
{"foo_d", :deleted},
152
-
], more: false} = MemKV.scan(kv, 3, "foo_b", "foo_e", limit: 3)
152
+
], count: 2, range: {"foo_b", "foo_e"}, more: false} = MemKV.scan(kv, 3, "foo_b", "foo_e", limit: 3)
153
153
end
154
154
155
155
test "scans backward", %{kv: kv} do
···
158
158
{"foo_c", "bar_c"},
159
159
{"foo_b", "bar_b"},
160
160
{"foo_a", "bar_a"},
161
-
], more: false} = MemKV.scan(kv, 1, "foo_", "foo_e", reverse: true)
161
+
], count: 4, range: {"foo_", "foo_e"}, more: false} = MemKV.scan(kv, 1, "foo_", "foo_e", reverse: true)
162
162
163
163
assert %RangeResult{pairs: [
164
164
{"foo_d", "bar_d"},
165
165
{"foo_c", "bar_c"},
166
166
{"foo_b", "bar_b"},
167
167
{"foo_a", "bar_a_2"},
168
-
], more: false} = MemKV.scan(kv, 2, "foo_", "foo_e", reverse: true)
168
+
], count: 4, range: {"foo_", "foo_e"}, more: false} = MemKV.scan(kv, 2, "foo_", "foo_e", reverse: true)
169
169
170
170
assert %RangeResult{pairs: [
171
171
{"foo_e", "bar_e_3"},
···
173
173
{"foo_c", "bar_c"},
174
174
{"foo_b", "bar_b_3"},
175
175
{"foo_a", "bar_a_2"},
176
-
], more: false} = MemKV.scan(kv, 3, "foo_", "foo_f", reverse: true)
176
+
], count: 4, range: {"foo_", "foo_f"}, more: false} = MemKV.scan(kv, 3, "foo_", "foo_f", reverse: true)
177
177
end
178
178
179
179
test "limits backward", %{kv: kv} do
···
181
181
{"foo_d", :deleted},
182
182
{"foo_c", "bar_c"},
183
183
{"foo_b", "bar_b_3"},
184
-
], more: true} = MemKV.scan(kv, 3, "foo_", "foo_e", reverse: true, limit: 2)
184
+
], count: 2, range: {"foo_b", "foo_e"}, more: true} = MemKV.scan(kv, 3, "foo_", "foo_e", reverse: true, limit: 2)
185
185
186
186
assert %RangeResult{pairs: [
187
187
{"foo_d", :deleted},
188
188
{"foo_c", "bar_c"},
189
189
{"foo_b", "bar_b_3"},
190
-
], more: false} = MemKV.scan(kv, 3, "foo_b", "foo_e", reverse: true, limit: 3)
190
+
], count: 2, range: {"foo_b", "foo_e"}, more: false} = MemKV.scan(kv, 3, "foo_b", "foo_e", reverse: true, limit: 3)
191
191
end
192
192
end
193
193