rough initial version of merging

Author: ckampfe

lib/b4.ex

@@ -52,9 +52,15 @@ defmodule B4 do
   end
 
   def merge(directory, timeout \\ 5_000) do
-    :ok = Writer.set_merge_in_progress(directory, true)
-    :ok = KeydirOwner.merge(directory, timeout)
-    :ok = Writer.set_merge_in_progress(directory, false)
+    with {_, :ok} <- {:set_merge_in_progress, Writer.set_merge_in_progress(directory, true)},
+         {_, :ok} <- {:merge_action, KeydirOwner.merge(directory, timeout)},
+         {_, :ok} <- {:unset_merge_in_process, Writer.set_merge_in_progress(directory, false)} do
+      :ok
+    else
+      {_action, e} ->
+        Writer.set_merge_in_progress(directory, false)
+        {:error, e}
+    end
   end
 
   def close(directory) do

lib/b4/keydir.ex

@@ -28,6 +28,10 @@ defmodule B4.Keydir do
     end
   end
 
+  def contains_key?(tid, key) do
+    :ets.member(tid, key)
+  end
+
   def keys(tid) do
     tid
     |> :ets.match({:"$1", :_, :_, :_, :_})

lib/b4/keydir_owner.ex

@@ -6,8 +6,8 @@ defmodule B4.KeydirOwner do
   alias B4.{Files, Keydir, Writer}
 
   defmodule State do
-    @enforce_keys [:directory, :tid]
-    defstruct [:directory, :tid]
+    @enforce_keys [:directory, :tid, :target_file_size]
+    defstruct [:directory, :tid, :target_file_size]
   end
 
   def start_link(%{directory: directory} = args) do
@@ -23,7 +23,7 @@ defmodule B4.KeydirOwner do
   end
 
   @impl GenServer
-  def init(%{directory: directory} = _init_arg) do
+  def init(%{directory: directory, options: [target_file_size: target_file_size]} = _init_arg) do
     tid = Keydir.new()
 
     :ok = :persistent_term.put({:tid, directory}, tid)
@@ -34,12 +34,15 @@ defmodule B4.KeydirOwner do
       Files.apply_file_to_keydir(path, tid)
     end)
 
-    {:ok, %State{directory: directory, tid: tid}}
+    {:ok, %State{directory: directory, tid: tid, target_file_size: target_file_size}}
   end
 
   @impl GenServer
-  def handle_call(:merge, _from, %State{directory: directory} = state) do
-    # TODO do merge here
+  def handle_call(
+        :merge,
+        _from,
+        %State{directory: directory, tid: tid, target_file_size: target_file_size} = state
+      ) do
     # two sets:
     # old read file set
     # new read file set
@@ -51,20 +54,145 @@ defmodule B4.KeydirOwner do
     # ELSE
     #   skip
     # END
-    write_file_id = Writer.write_file_id(directory)
-
-    read_only_database_files = Files.read_only_database_files(directory, write_file_id)
-
-    Enum.each(read_only_database_files, fn path ->
-      nil
+    #
+    # at given any time, there can only be ONE live entry
+    # for a given key,
+    # meaning any other entries for a given key are
+    # therefor old, and free for deletion
+    current_write_file_id = Writer.write_file_id(directory)
+
+    read_only_database_files = Files.read_only_database_files(directory, current_write_file_id)
+
+    {:ok, %{write_file: merge_write_file, file_id: merge_write_file_id}} =
+      Writer.new_write_file(directory)
+
+    Enum.reduce(read_only_database_files, %{merge_file_ids: MapSet.new()}, fn path, outer_acc ->
+      acc_for_file =
+        path
+        |> Files.stream_entries()
+        |> Enum.reduce(
+          %{
+            merge_write_file: merge_write_file,
+            merge_write_file_id: merge_write_file_id,
+            merge_file_ids: MapSet.new(),
+            merge_write_file_position: 0
+          },
+          fn %{
+               entry: %{
+                 crc32: crc32,
+                 entry_id: on_disk_entry_id,
+                 key_size: key_size,
+                 value_size: value_size,
+                 key_bytes: key_bytes,
+                 value_bytes: value_bytes
+               },
+               meta: %{}
+             },
+             %{
+               merge_write_file: merge_write_file,
+               merge_write_file_id: merge_write_file_id,
+               merge_write_file_position: merge_write_file_position
+             } =
+               acc ->
+            key = :erlang.binary_to_term(key_bytes)
+
+            case Keydir.fetch(tid, key) do
+              {:ok, {_key, _file_id, _entry_size, _file_position, keydir_entry_id}}
+              when keydir_entry_id == on_disk_entry_id ->
+                {:ok,
+                 %{
+                   merge_write_file: merge_write_file,
+                   merge_write_file_id: merge_write_file_id,
+                   merge_write_file_position: merge_write_file_position
+                 }} =
+                  if merge_write_file_position >= target_file_size do
+                    {:ok, %{write_file: merge_write_file, file_id: merge_write_file_id}} =
+                      Writer.new_write_file(directory)
+
+                    {:ok,
+                     %{
+                       merge_write_file: merge_write_file,
+                       merge_write_file_id: merge_write_file_id,
+                       merge_write_file_position: 0
+                     }}
+                  else
+                    {:ok,
+                     %{
+                       merge_write_file: merge_write_file,
+                       merge_write_file_id: merge_write_file_id,
+                       merge_write_file_position: merge_write_file_position
+                     }}
+                  end
+
+                entry =
+                  [
+                    Writer.int_to_u32_bytes(crc32),
+                    Writer.int_to_u128_bytes(on_disk_entry_id),
+                    Writer.int_to_u32_bytes(key_size),
+                    Writer.int_to_u32_bytes(value_size),
+                    key_bytes,
+                    value_bytes
+                  ]
+
+                :ok = :file.write(merge_write_file, entry)
+
+                entry_size = :erlang.iolist_size(entry)
+
+                true =
+                  Keydir.insert(
+                    tid,
+                    key,
+                    merge_write_file_id,
+                    entry_size,
+                    merge_write_file_position,
+                    on_disk_entry_id
+                  )
+
+                %{
+                  acc
+                  | merge_write_file: merge_write_file,
+                    merge_write_file_id: merge_write_file_id,
+                    merge_write_file_position: acc.merge_write_file_position + entry_size,
+                    merge_file_ids: MapSet.put(acc.merge_file_ids, merge_write_file_id)
+                }
+
+              # the entry isn't in the keydir,
+              # so it isn't live anymore,
+              # so skip it
+              :error ->
+                acc
+
+              # the ids for the given key didn't match,
+              # so they are old version of that key,
+              # so we ignore them
+              _ ->
+                acc
+            end
+          end
+        )
+
+      Map.update!(outer_acc, :merge_file_ids, fn merge_file_ids ->
+        MapSet.union(merge_file_ids, acc_for_file[:merge_file_ids])
+      end)
     end)
 
-    # - read every entry in the read_only_database_files
-    # - if the key is in keydir and the id == current id:
-    #     keep the key, add to new read file set, update keydir
-    #   else
-    #     skip
-    #   end
+    # |> IO.inspect(label: "merge file acc state")
+
+    # deleting these files happens AFTER
+    # the keydir has been updated with the new location
+    # on disk for the given entry,
+    # so this data is "dead",
+    # and there are no readers of it,
+    # barring some reader that has initiated some incredibly
+    # slow read prior to the merge starting.
+    #
+    # TODO
+    # this is a possibility, but not a huge one,
+    # so we should probably guard for this in the future in some way.
+    Enum.each(read_only_database_files, fn old_read_only_file ->
+      File.rm!(old_read_only_file)
+      # IO.inspect("deleted #{old_read_only_file}")
+    end)
 
     {:reply, :ok, state}
   end

lib/b4/writer.ex

@@ -199,6 +199,10 @@ defmodule B4.Writer do
     <<integer::unsigned-integer-32>>
   end
 
+  def int_to_u128_bytes(integer) when is_integer(integer) do
+    <<integer::unsigned-integer-128>>
+  end
+
   def name(directory) do
     :"#{__MODULE__}-#{directory}"
   end

test/b4_test.exs

@@ -121,4 +121,88 @@ defmodule B4Test do
     assert :ok = B4.delete(dir, "c")
     assert [] = Enum.sort(B4.keys(dir))
   end
+
+  test "simple merge", %{dir: dir} do
+    Enum.each(1..4, fn i ->
+      assert :ok = B4.new(dir)
+      assert :ok = B4.insert(dir, "a", i)
+      assert :ok = B4.close(dir)
+    end)
+
+    # File.ls!(dir) |> IO.inspect(label: "preexisting read files")
+    assert Enum.count(File.ls!(dir)) == 4
+
+    # Enum.each(File.ls!(dir), fn file ->
+    #   Path.join([dir, file])
+    #   |> File.stat!()
+    #   |> IO.inspect()
+    # end)
+
+    assert :ok = B4.new(dir)
+    # one more file because opening the db always creates a fresh writer file
+    assert Enum.count(File.ls!(dir)) == 5
+    # File.ls!(dir) |> IO.inspect(label: "all files pre merge")
+    assert :ok = B4.merge(dir)
+    # merging should:
+    # - create 1 new merge file (for this particular dataset)
+    # - delete the 4 previous read files
+    # - leave the current write file untouched
+    # for a total of 2 files
+    assert Enum.count(File.ls!(dir)) == 2
+    # File.ls!(dir) |> IO.inspect(label: "all files post merge")
+    assert :ok = B4.close(dir)
+
+    assert :ok = B4.new(dir)
+    # we should now have:
+    # - the new current write file
+    # - the previous write file
+    # - the merge file
+    assert Enum.count(File.ls!(dir)) == 3
+    assert ["a"] = B4.keys(dir)
+    assert {:ok, 4} = B4.fetch(dir, "a")
+  end
+
+  test "simple merge with deletes", %{dir: dir} do
+    Enum.each(1..4, fn i ->
+      assert :ok = B4.new(dir)
+      assert :ok = B4.insert(dir, "a", i)
+      assert :ok = B4.insert(dir, "b", i)
+      assert :ok = B4.close(dir)
+    end)
+
+    # File.ls!(dir) |> IO.inspect(label: "preexisting read files")
+    assert Enum.count(File.ls!(dir)) == 4
+
+    # Enum.each(File.ls!(dir), fn file ->
+    #   Path.join([dir, file])
+    #   |> File.stat!()
+    #   |> IO.inspect()
+    # end)
+
+    assert :ok = B4.new(dir)
+    assert :ok = B4.delete(dir, "a")
+
+    # one more file because opening the db always creates a fresh writer file
+    assert Enum.count(File.ls!(dir)) == 5
+    # File.ls!(dir) |> IO.inspect(label: "all files pre merge")
+    assert :ok = B4.merge(dir)
+    # merging should:
+    # - create 1 new merge file (for this particular dataset)
+    # - delete the 4 previous read files
+    # - leave the current write file untouched
+    # for a total of 2 files
+    assert Enum.count(File.ls!(dir)) == 2
+    # File.ls!(dir) |> IO.inspect(label: "all files post merge")
+    assert :ok = B4.close(dir)
+
+    assert :ok = B4.new(dir)
+    # we should now have:
+    # - the new current write file
+    # - the previous write file
+    # - the merge file
+    assert Enum.count(File.ls!(dir)) == 3
+    assert ["b"] = B4.keys(dir)
+    assert :not_found = B4.fetch(dir, "a")
+    assert {:ok, 4} = B4.fetch(dir, "b")
+  end
 end