Always write a bloom filter, even when empty, just to avoid over complicating

things. Reuse the incremental merge trigger when closing the nursery file.

src/hanoidb_merger.erl

@@ -52,8 +52,16 @@ merge(A,B,C, Size, IsLastLevel, Options) ->
             false ->
                 hanoidb_writer:open(C, [{size, Size} | Options])
         end,
-    {node, AKVs} = hanoidb_reader:first_node(BT1),
-    {node, BKVs} = hanoidb_reader:first_node(BT2),
+    AKVs =
+        case hanoidb_reader:first_node(BT1) of
+            {node, AKV} -> AKV;
+            none -> []
+        end,
+    BKVs =
+        case hanoidb_reader:first_node(BT2) of
+            {node, BKV} ->BKV;
+            none -> []
+        end,
     scan(BT1, BT2, Out, IsLastLevel, AKVs, BKVs, {0, none}).
 
 terminate(Out) ->

src/hanoidb_nursery.erl

@@ -177,15 +177,15 @@ lookup(Key, #nursery{cache=Cache}) ->
 %% Finish this nursery (encode it to a btree, and delete the nursery file)
 %% @end
 -spec finish(Nursery::#nursery{}, TopLevel::pid()) -> ok.
-finish(#nursery{ dir=Dir, cache=Cache, log_file=LogFile, count=Count,
-                 config=Config, merge_done=DoneMerge }, TopLevel) ->
+finish(#nursery{ dir=Dir, cache=Cache, log_file=LogFile,
+                 count=Count, config=Config }=Nursery, TopLevel) ->
 
     hanoidb_util:ensure_expiry(Config),
 
     %% First, close the log file (if it is open)
-    case LogFile /= undefined of
-        true -> ok = file:close(LogFile);
-        false -> ok
+    case LogFile of
+        undefined -> ok;
+        _ -> ok = file:close(LogFile)
     end,
 
     case Count of
@@ -203,16 +203,12 @@ finish(#nursery{ dir=Dir, cache=Cache, log_file=LogFile, count=Count,
                 ok = hanoidb_writer:close(BT)
             end,
 
-            %% inject the B-Tree (blocking RPC)
+            %% Inject the B-Tree (blocking RPC)
             ok = hanoidb_level:inject(TopLevel, BTreeFileName),
 
-            %% issue some work if this is a top-level inject (blocks until previous such
+            %% Issue some work if this is a top-level inject (blocks until previous such
             %% incremental merge is finished).
-            if DoneMerge >= ?BTREE_SIZE(?TOP_LEVEL) ->
-                    ok;
-               true ->
-                    hanoidb_level:begin_incremental_merge(TopLevel, ?BTREE_SIZE(?TOP_LEVEL) - DoneMerge)
-            end;
+            {ok, Nursery2} = do_inc_merge(Nursery, Count, TopLevel);
 
         _ ->
             ok
@@ -285,11 +281,10 @@ transact(Spec, Nursery=#nursery{ log_file=File, cache=Cache0, total_size=TotalSi
     do_inc_merge(Nursery2#nursery{ cache=Cache2, total_size=TotalSize+erlang:iolist_size(Data), count=Count }, length(Spec), Top).
 
 do_inc_merge(Nursery=#nursery{ step=Step, merge_done=Done }, N, TopLevel) ->
-    case Step+N >= ?INC_MERGE_STEP of
-        true ->
+    if Step+N >= ?INC_MERGE_STEP ->
             hanoidb_level:begin_incremental_merge(TopLevel, Step + N),
             {ok, Nursery#nursery{ step=0, merge_done=Done + Step + N }};
-        false ->
+       true ->
             {ok, Nursery#nursery{ step=Step + N }}
     end.
 

src/hanoidb_reader.erl

@@ -80,16 +80,8 @@ open(Name, Config) ->
             %% read root position
             {ok, <<RootPos:64/unsigned>>} = file:pread(File, FileInfo#file_info.size - 8, 8),
             {ok, <<BloomSize:32/unsigned>>} = file:pread(File, FileInfo#file_info.size - 12, 4),
-
-            Bloom =
-                case BloomSize of
-                    0 ->
-                        {ok, <<NumElems:32/unsigned>>} = file:pread(File, (FileInfo#file_info.size - 16), 4),
-                        bloom:bloom(NumElems);
-                    _ ->
             {ok, BloomData} = file:pread(File, (FileInfo#file_info.size - 12 - BloomSize), BloomSize),
-                        bloom:decode(BloomData)
-                end,
+            Bloom = bloom:decode(BloomData),
 
             %% read in the root node
             Root =

src/hanoidb_writer.erl

@@ -188,17 +188,17 @@ do_open(Name, Options, OpenOpts) ->
 %% @doc flush pending nodes and write trailer
 archive_nodes(#state{ nodes=[], last_node_pos=LastNodePos, last_node_size=_LastNodeSize, bloom=Bloom, index_file=IdxFile }=State) ->
 
-    {BloomBin, BloomSize, RootPos} =
+    BloomBin = bloom:encode(Bloom),
+    BloomSize = byte_size(BloomBin),
+    RootPos =
         case LastNodePos of
             undefined ->
                 %% store contains no entries
                 ok = file:write(IdxFile, <<0:32/unsigned, 0:16/unsigned>>),
-                {<<(bloom:size(Bloom)):32/unsigned>>, 0, ?FIRST_BLOCK_POS};
+                ?FIRST_BLOCK_POS;
             _ ->
-                EncodedBloom = bloom:encode(Bloom),
-                {EncodedBloom, byte_size(EncodedBloom), LastNodePos}
+                LastNodePos
         end,
-
     Trailer = << 0:32/unsigned, BloomBin/binary, BloomSize:32/unsigned,  RootPos:64/unsigned >>,
 
     ok = file:write(IdxFile, Trailer),