Support flush for range deletes in HybridKV · corporate.fm/hobbes@68dde3a

+7 -1

ROADMAP.md

··· 59 59 - [ ] Add include_start/end to Transaction.read_range 60 60 - [ ] Clears 61 61 - [X] Implement clears in KVs 62 - - [ ] Implement clears in database 62 + - [ ] Implement clears in the database 63 63 - [ ] Range clears 64 + - [ ] Implement range clears in KVs 65 + - [X] Add persistent tree (RangeForest) to hold deleted ranges 66 + - [X] Handle deleted ranges in HybridKV.get/put 67 + - [X] Handle deleted ranges in HybridKV.flush 68 + - [ ] Handle deleted ranges in HybridKV.scan 69 + - [ ] Implement range clears in the database 64 70 - [ ] Recovery 65 71 - [ ] Monitor all transaction processes in Manager to detect failures 66 72 - [ ] Lock TLogs and choose recovery version

+59 -21

lib/hybrid_kv.ex

··· 1 1 defmodule Hobbes.HybridKV do 2 - alias Hobbes.{HybridKV, MemKV, RangeForest} 2 + alias Hobbes.{HybridKV, MemKV, RangeForest, Utils} 3 3 alias Hobbes.RangeForest.RangeTree 4 4 alias Hobbes.KV.FlatKV 5 5 alias Hobbes.Structs.RangeResult 6 6 7 - @enforce_keys [:mem_kv, :storage_kv, :deleted_forest] 7 + @enforce_keys [:mem_kv, :storage_kv, :deleted_forest, :flushed_version] 8 8 defstruct @enforce_keys 9 9 10 10 def new do ··· 12 12 mem_kv: MemKV.new(), 13 13 storage_kv: FlatKV.new(), 14 14 deleted_forest: RangeForest.new(), 15 + flushed_version: 0, 15 16 } 16 17 end 17 18 ··· 39 40 @spec get(%HybridKV{}, non_neg_integer, binary) :: binary | nil 40 41 def get(%HybridKV{} = kv, version, key) 41 42 when is_integer(version) and version >= 0 and is_binary(key) do 42 - floor_version = 43 - case RangeForest.tree_at(kv.deleted_forest, version) |> RangeTree.intersect_key(0, key) do 44 - {_sk, _ek, v} -> v + 1 45 - nil -> 0 46 - end 43 + kv.deleted_forest 44 + |> RangeForest.tree_at(version) 45 + |> RangeTree.intersect_key(kv.flushed_version + 1, key) 46 + |> case do 47 + nil -> 48 + # There are no overlapping range deletes 49 + case MemKV.get(kv.mem_kv, version, 0, key) do 50 + nil -> FlatKV.get(kv.storage_kv, key) 51 + :deleted -> nil 52 + value -> value 53 + end 47 54 48 - case MemKV.get(kv.mem_kv, version, floor_version, key) do 49 - nil -> FlatKV.get(kv.storage_kv, key) 50 - :deleted -> nil 51 - value -> value 55 + {_sk, _ek, range_deleted_at} -> 56 + # There is an overlapping range delete, so we return nil if 57 + # a newer key is not found 58 + case MemKV.get(kv.mem_kv, version, range_deleted_at + 1, key) do 59 + nil -> nil 60 + :deleted -> nil 61 + value -> value 62 + end 52 63 end 53 64 end 54 65 ··· 198 209 @doc """ 199 210 Flushes keys/values with a version <= `version` to unversioned storage. 200 211 """ 201 - @spec flush(%HybridKV{}, non_neg_integer) :: :ok 212 + @spec flush(%HybridKV{}, non_neg_integer) :: %HybridKV{} 202 213 def flush(%HybridKV{} = kv, version) when is_integer(version) and version >= 0 do 203 - MemKV.flush(kv.mem_kv, version) 204 - # Sort here is not needed for determinism, but the btree 205 - # inserts will probably prefer sorted data in the future 206 - # TODO: maybe return sorted keys from MemKV.flush? 207 - |> Enum.sort_by(&(&1)) 208 - |> Enum.map(fn 209 - {k, :deleted} -> FlatKV.delete(kv.storage_kv, k) 210 - {k, v} -> FlatKV.put(kv.storage_kv, k, v) 214 + {ranges, deleted_forest} = RangeForest.flush(kv.deleted_forest, version, kv.flushed_version + 1) 215 + 216 + # Gather all mutations (writes, deletes, and range deletes) and batch them together 217 + # by version 218 + ranges_per_version = 219 + Enum.group_by(ranges, fn {_start, _end, v} -> v end) 220 + |> Enum.sort_by(&elem(&1, 0)) 221 + 222 + pairs_per_version = 223 + MemKV.flush(kv.mem_kv, version) 224 + |> Enum.group_by(fn {_key, _value, v} -> v end) 225 + |> Enum.sort_by(&elem(&1, 0)) 226 + 227 + batches = Utils.interleave_batches(pairs_per_version, ranges_per_version) 228 + 229 + storage_kv = kv.storage_kv 230 + # Iterate through all batches and apply point writes first, 231 + # then range deletes 232 + # 233 + # Note that position of the range deletes within the batch does 234 + # not matter because we split them for writes which come later 235 + # in the same version (in put/4) 236 + # 237 + # TODO: we should probably switch to a "mutation log" approach 238 + # to reduce complexity (performance would probably be better anyway) 239 + Enum.each(batches, fn {_v, pairs, ranges} -> 240 + Enum.each(pairs, fn 241 + {key, :deleted, _v} -> FlatKV.delete(storage_kv, key) 242 + {key, value, _v} -> FlatKV.put(storage_kv, key, value) 243 + end) 244 + 245 + Enum.each(ranges, fn {sk, ek, _v} -> 246 + FlatKV.clear_range(storage_kv, sk, ek) 247 + end) 211 248 end) 212 - :ok 249 + 250 + %HybridKV{kv | deleted_forest: deleted_forest, flushed_version: version} 213 251 end 214 252 215 253 @spec put_storage(%HybridKV{}, binary, binary) :: :ok

+9 -3

lib/mem_kv.ex

··· 208 208 } 209 209 210 210 """ 211 - @spec flush(:ets.table, non_neg_integer) :: %{String.t => String.t} 211 + @spec flush(:ets.table, non_neg_integer) :: [{binary, binary, non_neg_integer}] 212 212 def flush(table, version) do 213 213 # TODO: since this scans the entire table anyway, it can be rewritten 214 214 # to be much faster with matchspecs 215 - do_flush_scan(table, version, {"", -1}, %{}) 215 + do_flush_scan(table, version, {"", -1}, []) 216 + |> Enum.reverse() 216 217 end 217 218 218 219 defp do_flush_scan(table, version, prev, acc) do ··· 223 224 [{^full_key, value}] = :ets.lookup(table, full_key) 224 225 true = :ets.delete(table, full_key) 225 226 226 - do_flush_scan(table, version, full_key, Map.put(acc, key, value)) 227 + case acc do 228 + [{^key, _value, _v} | rest] -> 229 + do_flush_scan(table, version, full_key, [{key, value, ver} | rest]) 230 + acc -> 231 + do_flush_scan(table, version, full_key, [{key, value, ver} | acc]) 232 + end 227 233 228 234 false -> 229 235 do_flush_scan(table, version, full_key, acc)

+7

lib/range_forest.ex

··· 162 162 # tree to the end of the queue 163 163 tree = RangeTree.insert_range(prev_tree, version, start_key, end_key) 164 164 :gb_trees.insert(version, tree, forest) 165 + 166 + :none -> 167 + # There is no previous tree, so we create a new one 168 + tree = RangeTree.new() |> RangeTree.insert_range(version, start_key, end_key) 169 + :gb_trees.insert(version, tree, forest) 165 170 end 166 171 end 167 172 ··· 184 189 def tree_at(forest, version) do 185 190 case :gb_trees.smaller(version + 1, forest) do 186 191 {_v, tree} -> tree 192 + :none -> :gb_trees.empty() 187 193 end 188 194 end 189 195 ··· 195 201 {[], forest} 196 202 197 203 {last_tree, forest} -> 204 + # TODO: clear these ranges from the newest tree 198 205 # TODO: should end_key be four \xFFs to include special space? 199 206 ranges = RangeTree.intersect_range(last_tree, min_version, "", "\xFF\xFF") 200 207 {ranges, forest}

+21 -10

test/hybrid_kv_test.exs

··· 47 47 e in [ExUnit.AssertionError] -> 48 48 e = Map.update!(e, :message, &(&1 <> " (at op #{i}, seed=#{inspect(verifier.seed)})")) 49 49 reraise e, __STACKTRACE__ 50 + e -> 51 + require Logger 52 + Logger.error("Error #{inspect(e)} at op=#{i}, seed=#{inspect(verifier.seed)}") 53 + reraise e, __STACKTRACE__ 50 54 end 51 55 end) 52 56 end ··· 56 60 dbg [ 57 61 hybrid_kv: HybridKV.dump(verifier.hybrid_kv), 58 62 test_kv: TestKV.dump(verifier.test_kv), 63 + durable_version: verifier.durable_version, 59 64 ], limit: :infinity 60 65 end 61 66 62 67 @ops [:put, :delete, :delete_range, :get] 63 68 defp random_op do 64 69 case Enum.random(1..100) do 65 - #1 -> :flush 70 + 1 -> :flush 66 71 _ -> Enum.random(@ops) 67 72 end 68 73 end ··· 147 152 148 153 defp perform(:flush, %Verifier{hybrid_kv: hkv} = verifier) do 149 154 flush_version = rand(verifier.durable_version, verifier.version) 150 - :ok = HybridKV.flush(hkv, flush_version) 155 + hkv = HybridKV.flush(hkv, flush_version) 151 156 152 - %Verifier{verifier | durable_version: flush_version} 157 + %Verifier{verifier | hybrid_kv: hkv, durable_version: flush_version} 153 158 end 154 159 155 160 defp inc_version(%Verifier{version: version} = verifier) do ··· 188 193 describe "verify HybridKV" do 189 194 @tag :hkv_verify 190 195 test "operations" do 191 - Verifier.new({101, 101, 102}) 196 + Verifier.new({100, 101, 102}) 192 197 |> Verifier.run(1000) 193 198 end 194 199 ··· 224 229 assert HybridKV.get(kv, 2, "foo") == "bar_2" 225 230 assert HybridKV.get(kv, 3, "foo") == "bar_3" 226 231 227 - assert :ok = HybridKV.flush(kv, 2) 232 + assert kv = HybridKV.flush(kv, 2) 228 233 229 234 assert HybridKV.get(kv, 1, "foo") == "bar_2" 230 235 assert HybridKV.get(kv, 2, "foo") == "bar_2" ··· 239 244 {"foo", "bar"}, 240 245 ], more: false} = HybridKV.scan(kv, 1, "foo", "foo_z") 241 246 242 - assert :ok = HybridKV.flush(kv, 2) 247 + assert kv = HybridKV.flush(kv, 2) 243 248 244 249 assert %RangeResult{pairs: [ 245 250 {"foo", "bar_2"}, ··· 264 269 {"foo_c", "bar_c_3"}, 265 270 ], more: false} = HybridKV.scan(kv, 4, "foo", "foo_z") 266 271 267 - :ok = HybridKV.flush(kv, 4) 272 + kv = HybridKV.flush(kv, 4) 268 273 269 274 assert %RangeResult{pairs: [ 270 275 {"foo", "bar_3"}, ··· 274 279 end 275 280 276 281 test "scan limit", %{kv: kv} do 277 - assert :ok = HybridKV.flush(kv, 2) 282 + assert kv = HybridKV.flush(kv, 2) 278 283 279 284 assert %RangeResult{pairs: [ 280 285 {"foo", "bar_3"}, ··· 284 289 285 290 test "scan deleted limit", %{kv: kv} do 286 291 Enum.each(0..5, fn ver -> 287 - if ver != 0, do: HybridKV.flush(kv, ver) 292 + kv = case ver == 0 do 293 + true -> HybridKV.flush(kv, ver) 294 + false -> kv 295 + end 288 296 289 297 assert %RangeResult{pairs: [ 290 298 {"foo", "bar_3"}, ··· 299 307 HybridKV.put(kv, 5, "foo_f", "bar_f_5") 300 308 301 309 Enum.each(0..5, fn ver -> 302 - if ver != 0, do: HybridKV.flush(kv, ver) 310 + kv = case ver == 0 do 311 + true -> HybridKV.flush(kv, ver) 312 + false -> kv 313 + end 303 314 304 315 assert %RangeResult{pairs: [ 305 316 {"foo", "bar_3"},

+10 -8

test/mem_kv_test.exs

··· 199 199 200 200 MemKV.put(kv, 2, "foo_b", "bar_b_2") 201 201 MemKV.put(kv, 2, "foo_c", "bar_c_2") 202 + MemKV.put(kv, 1, "foo_d", "bar_d") 202 203 203 204 assert MemKV.get(kv, 1, 0, "foo") == "bar" 204 205 assert MemKV.get(kv, 3, 0, "foo") == "bar_3" 205 206 206 - assert MemKV.flush(kv, 2) == %{ 207 - "foo" => "bar_2", 208 - "foo_b" => "bar_b_2", 209 - "foo_c" => "bar_c_2", 210 - } 207 + assert MemKV.flush(kv, 2) == [ 208 + {"foo", "bar_2", 2}, 209 + {"foo_b", "bar_b_2", 2}, 210 + {"foo_c", "bar_c_2", 2}, 211 + {"foo_d", "bar_d", 1}, 212 + ] 211 213 212 214 assert MemKV.get(kv, 1, 0, "foo") == nil 213 215 assert MemKV.get(kv, 2, 0, "foo") == nil ··· 221 223 MemKV.put(kv, 1, "foo", "bar") 222 224 MemKV.put(kv, 1, "\xFF\xFFhello", "world") 223 225 224 - assert MemKV.flush(kv, 1) == %{ 225 - "foo" => "bar", 226 - } 226 + assert MemKV.flush(kv, 1) == [ 227 + {"foo", "bar", 1}, 228 + ] 227 229 228 230 assert MemKV.get(kv, 1, 0, "foo") == nil 229 231 assert MemKV.get(kv, 1, 0, "\xFF\xFFhello") == "world"

Configure Feed

Configure Feed