Mas d34 leveled.i459 partialmerge (#460)

* Add test to replicate issue 459 Nothing actually crashes due to the issue - but looking at the logs there is the polarised stats associated with the issue. When merging into L3, you would normally expect to merge into 4 files - but actually we see FileCounter occasionally spiking. * Add partial merge support There is a `max_mergebelow` size which can be a positive integer, or infinity. It defaults to 32. If a merge from Level N covers less than `max_mergebelow` files in level N + 1 - the merge will proceesd as before. If it has >= `max_mergebelow`, the merge will be curtailed when `max_mergebelow div 2` files have been created at that level. The remainder for Level N will then be written, as well as for Level N + 1 up to the next whole file that has no yet been touched by the merge. The backlog that prompted the merge will still exist - as the files in Level N have not been changed. However, it is likely the next file picked will not be the same one, and will in probability have a lower number of files to merge (as the average is =< 8). This will stop progress from being halted by long merge jobs, as they will exit out in a safe way after partial completion. In the case where the majority of files covered do not require a merge, then those files will be skipped the next time the remainder file is picked up for merge at Level N
martinsumner · Nov 30, 2024 · 159425c · 159425c
1 parent 051aa12
commit 159425c
Show file tree

Hide file tree

Showing 8 changed files with 387 additions and 76 deletions.
diff --git a/include/leveled.hrl b/include/leveled.hrl
@@ -33,6 +33,7 @@
 %%% Non-configurable startup defaults
 %%%============================================================================
 -define(MAX_SSTSLOTS, 256).
+-define(MAX_MERGEBELOW, 24).
 -define(LOADING_PAUSE, 1000).
 -define(LOADING_BATCH, 1000).
 -define(CACHE_SIZE_JITTER, 25).
@@ -107,7 +108,8 @@
                         press_level = ?COMPRESSION_LEVEL :: non_neg_integer(),
                         log_options = leveled_log:get_opts() 
                             :: leveled_log:log_options(),
-                        max_sstslots = ?MAX_SSTSLOTS :: pos_integer(),
+                        max_sstslots = ?MAX_SSTSLOTS :: pos_integer()|infinity,
+                        max_mergebelow = ?MAX_MERGEBELOW :: pos_integer()|infinity,
                         pagecache_level = ?SST_PAGECACHELEVEL_NOLOOKUP
                             :: pos_integer(),
                         monitor = {no_monitor, 0}

diff --git a/rebar.config b/rebar.config
@@ -29,8 +29,8 @@
  ]}.
 
 {deps, [
-  {lz4, ".*", {git, "https://github.com/nhs-riak/erlang-lz4", {branch, "nhse-develop"}}},
-  {zstd, ".*", {git, "https://github.com/nhs-riak/zstd-erlang", {branch, "nhse-develop"}}}
+  {lz4, ".*", {git, "https://github.com/OpenRiak/erlang-lz4", {branch, "openriak-3.2"}}},
+  {zstd, ".*", {git, "https://github.com/OpenRiak/zstd-erlang", {branch, "openriak-3.2"}}}
         ]}.
 
 {ct_opts, [{dir, ["test/end_to_end"]}]}.
diff --git a/src/leveled_bookie.erl b/src/leveled_bookie.erl
@@ -114,6 +114,7 @@
                 {max_journalsize, 1000000000},
                 {max_journalobjectcount, 200000},
                 {max_sstslots, 256},
+                {max_mergebelow, 24},
                 {sync_strategy, ?DEFAULT_SYNC_STRATEGY},
                 {head_only, false},
                 {waste_retention_period, undefined},
@@ -201,6 +202,12 @@
             % The maximum number of slots in a SST file.  All testing is done
             % at a size of 256 (except for Quickcheck tests}, altering this
             % value is not recommended
+        {max_mergeblow, pos_integer()|infinity} |
+            % The maximum number of files for a single file to be merged into
+            % within the ledger.  If less than this, the merge will continue
+            % without a maximum.  If this or more overlapping below, only up
+            % to max_mergebelow div 2 additions should be created (the merge
+            % should be partial)
         {sync_strategy, sync_mode()} |
             % Should be sync if it is necessary to flush to disk after every
             % write, or none if not (allow the OS to schecdule).  This has a
@@ -293,7 +300,7 @@
             % To which level of the ledger should the ledger contents be
             % pre-loaded into the pagecache (using fadvise on creation and
             % startup)
-            {compression_method, native|lz4|zstd|none} |
+        {compression_method, native|lz4|zstd|none} |
             % Compression method and point allow Leveled to be switched from
             % using bif based compression (zlib) to using nif based compression
             % (lz4 or zstd).
@@ -1836,6 +1843,7 @@ set_options(Opts, Monitor) ->
     CompressionLevel = proplists:get_value(compression_level, Opts),
 
     MaxSSTSlots = proplists:get_value(max_sstslots, Opts),
+    MaxMergeBelow = proplists:get_value(max_mergebelow, Opts),
 
     ScoreOneIn = proplists:get_value(journalcompaction_scoreonein, Opts),
 
@@ -1869,6 +1877,7 @@ set_options(Opts, Monitor) ->
                                     press_level = CompressionLevel,
                                     log_options = leveled_log:get_opts(),
                                     max_sstslots = MaxSSTSlots,
+                                    max_mergebelow = MaxMergeBelow,
                                     monitor = Monitor},
                             monitor = Monitor}
         }.

diff --git a/src/leveled_log.erl b/src/leveled_log.erl
@@ -147,7 +147,7 @@
         pc010 =>
             {info, <<"Merge to be commenced for FileToMerge=~s with MSN=~w">>},
         pc011 =>
-            {info, <<"Merge completed with MSN=~w to Level=~w and FileCounter=~w">>},
+            {info, <<"Merge completed with MSN=~w to Level=~w and FileCounter=~w merge_type=~w">>},
         pc012 =>
             {debug, <<"File to be created as part of MSN=~w Filename=~s IsBasement=~w">>},
         pc013 =>
@@ -172,6 +172,8 @@
             {info, <<"Grooming compaction picked file with tomb_count=~w">>},
         pc025 =>
             {info, <<"At level=~w file_count=~w average words for heap_block_size=~w heap_size=~w recent_size=~w bin_vheap_size=~w">>},
+        pc026 =>
+            {info, <<"Performing potential partial to level=~w merge as FileCounter=~w restricting to MaxAdditions=~w">>},
         pm002 =>
             {info, <<"Completed dump of L0 cache to list of l0cache_size=~w">>},
         sst03 =>

diff --git a/src/leveled_pclerk.erl b/src/leveled_pclerk.erl
@@ -262,61 +262,177 @@ perform_merge(Manifest,
     MaxSQN = leveled_sst:sst_getmaxsequencenumber(Src#manifest_entry.owner),
     SinkLevel = SrcLevel + 1,
     SinkBasement = leveled_pmanifest:is_basement(Manifest, SinkLevel),
-    Additions = 
-        do_merge(SrcList, SinkList,
-                    SinkLevel, SinkBasement,
-                    RootPath, NewSQN, MaxSQN,
-                    OptsSST,
-                    []),
-    RevertPointerFun =
-        fun({next, ME, _SK}) ->
-            ME
+    MaxMergeBelow = OptsSST#sst_options.max_mergebelow,
+    MergeLimit = merge_limit(SrcLevel, length(SinkList), MaxMergeBelow),
+    {L2Additions, L1Additions, L2FileRemainder} = 
+        do_merge(
+            SrcList, SinkList,
+            SinkLevel, SinkBasement,
+            RootPath, NewSQN, MaxSQN,
+            OptsSST,
+            [],
+            MergeLimit
+        ),
+    RevertPointerFun = fun({next, ME, _SK}) -> ME end,
+    SinkManifestRemovals =
+        lists:subtract(
+            lists:map(RevertPointerFun, SinkList),
+            lists:map(RevertPointerFun, L2FileRemainder)
+        ),
+    Man0 =
+        leveled_pmanifest:replace_manifest_entry(
+            Manifest,
+            NewSQN,
+            SinkLevel,
+            SinkManifestRemovals,
+            L2Additions
+        ),
+    Man1 =
+        case L1Additions of
+            [] ->
+                leveled_pmanifest:remove_manifest_entry(
+                    Man0,
+                    NewSQN,
+                    SrcLevel,
+                    Src
+                );
+            PartialFiles ->
+                leveled_pmanifest:replace_manifest_entry(
+                    Man0,
+                    NewSQN,
+                    SrcLevel,
+                    [Src],
+                    PartialFiles
+                )
         end,
-    SinkManifestList = lists:map(RevertPointerFun, SinkList),
-    Man0 = leveled_pmanifest:replace_manifest_entry(Manifest,
-                                                    NewSQN,
-                                                    SinkLevel,
-                                                    SinkManifestList,
-                                                    Additions),
-    Man2 = leveled_pmanifest:remove_manifest_entry(Man0,
-                                                    NewSQN,
-                                                    SrcLevel,
-                                                    Src),
-    {Man2, [Src|SinkManifestList]}.
-
-do_merge([], [], SinkLevel, _SinkB, _RP, NewSQN, _MaxSQN, _Opts, Additions) ->
-    leveled_log:log(pc011, [NewSQN, SinkLevel, length(Additions)]),
-    Additions;
-do_merge(KL1, KL2, SinkLevel, SinkB, RP, NewSQN, MaxSQN, OptsSST, Additions) ->
-    FileName = leveled_penciller:sst_filename(NewSQN,
-                                                SinkLevel,
-                                                length(Additions)),
+    {Man1, [Src|SinkManifestRemovals]}.
+
+-spec merge_limit(
+    non_neg_integer(), non_neg_integer(), pos_integer()|infinity)
+        -> pos_integer()|infinity.
+merge_limit(SrcLevel, SinkListLength, MMB) when SrcLevel =< 1; SinkListLength < MMB ->
+    infinity;
+merge_limit(SrcLevel, SinkListLength, MMB) when is_integer(MMB) ->
+    AdditionsLimit = max(1, MMB div 2),
+    leveled_log:log(pc026, [SrcLevel + 1, SinkListLength, AdditionsLimit]),
+    AdditionsLimit.
+
+-type merge_maybe_expanded_pointer() ::
+    leveled_codec:ledger_kv()|
+    leveled_sst:slot_pointer()|
+    leveled_sst:sst_pointer().
+        % Different to leveled_sst:maybe_expanded_pointer/0
+        % No sst_closed_pointer()
+
+do_merge(
+    [], [], SinkLevel, _SinkB, _RP, NewSQN, _MaxSQN, _Opts, Additions, _Max) ->
+    leveled_log:log(pc011, [NewSQN, SinkLevel, length(Additions), full]),
+    {lists:reverse(Additions), [], []};
+do_merge(
+    KL1, KL2, SinkLevel, SinkB, RP, NewSQN, MaxSQN, OptsSST, Additions, Max)
+        when length(Additions) >= Max ->
+    leveled_log:log(pc011, [NewSQN, SinkLevel, length(Additions), partial]),
+    FNSrc =
+        leveled_penciller:sst_filename(
+            NewSQN, SinkLevel - 1, 1
+        ),
+    FNSnk =
+        leveled_penciller:sst_filename(
+            NewSQN, SinkLevel, length(Additions) + 1
+        ),
+    {ExpandedKL1, []} = split_unexpanded_files(KL1),
+    {ExpandedKL2, L2FilePointersRem} = split_unexpanded_files(KL2),
+    TS1 = os:timestamp(),
+    InfOpts = OptsSST#sst_options{max_sstslots = infinity},
+        % Need to be careful to make sure all the remainder goes in one file,
+        % could be situations whereby the max_sstslots has been changed between
+        % restarts - and so there is too much data for one file in the
+        % remainder ... but don't want to loop round and consider more complex
+        % scenarios here.
+    NewMergeKL1 =
+        leveled_sst:sst_newmerge(
+            RP, FNSrc,ExpandedKL1, [], false, SinkLevel - 1, MaxSQN, InfOpts
+        ),
+    TS2 = os:timestamp(),
+    NewMergeKL2 =
+        leveled_sst:sst_newmerge(
+            RP, FNSnk, [], ExpandedKL2, SinkB, SinkLevel, MaxSQN, InfOpts
+        ),
+    {KL1Additions, [], []} = add_entry(NewMergeKL1, FNSrc, TS1, []),
+    {KL2Additions, [], []} = add_entry(NewMergeKL2, FNSnk, TS2, Additions),
+    {lists:reverse(KL2Additions), KL1Additions, L2FilePointersRem};
+do_merge(
+    KL1, KL2, SinkLevel, SinkB, RP, NewSQN, MaxSQN, OptsSST, Additions, Max) ->
+    FileName =
+        leveled_penciller:sst_filename(
+            NewSQN, SinkLevel, length(Additions)
+        ),
     leveled_log:log(pc012, [NewSQN, FileName, SinkB]),
     TS1 = os:timestamp(),
-    case leveled_sst:sst_newmerge(RP, FileName,
-                                    KL1, KL2, SinkB, SinkLevel, MaxSQN,
-                                    OptsSST) of
-        empty ->
-            leveled_log:log(pc013, [FileName]),
-            do_merge([], [],
-                        SinkLevel, SinkB,
-                        RP, NewSQN, MaxSQN,
-                        OptsSST, 
-                        Additions);                        
-        {ok, Pid, Reply, Bloom} ->
-            {{KL1Rem, KL2Rem}, SmallestKey, HighestKey} = Reply,
-                Entry = #manifest_entry{start_key=SmallestKey,
-                                            end_key=HighestKey,
-                                            owner=Pid,
-                                            filename=FileName,
-                                            bloom=Bloom},
-                leveled_log:log_timer(pc015, [], TS1),
-                do_merge(KL1Rem, KL2Rem,
-                            SinkLevel, SinkB,
-                            RP, NewSQN, MaxSQN,
-                            OptsSST,
-                            Additions ++ [Entry])
-    end.
+    NewMerge =
+        leveled_sst:sst_newmerge(
+            RP, FileName, KL1, KL2, SinkB, SinkLevel, MaxSQN, OptsSST),
+    {UpdAdditions, KL1Rem, KL2Rem} =
+        add_entry(NewMerge, FileName, TS1, Additions),
+    do_merge(
+        KL1Rem,
+        KL2Rem,
+        SinkLevel,
+        SinkB,
+        RP,
+        NewSQN,
+        MaxSQN,
+        OptsSST,
+        UpdAdditions,
+        Max
+    ).
+
+add_entry(empty, FileName, _TS1, Additions) ->
+    leveled_log:log(pc013, [FileName]),
+    {[], [], Additions};
+add_entry({ok, Pid, Reply, Bloom}, FileName, TS1, Additions) ->
+    {{KL1Rem, KL2Rem}, SmallestKey, HighestKey} = Reply,
+    Entry =
+        #manifest_entry{
+            start_key=SmallestKey,
+            end_key=HighestKey,
+            owner=Pid,
+            filename=FileName,
+            bloom=Bloom
+        },
+    leveled_log:log_timer(pc015, [], TS1),
+    {[Entry|Additions], KL1Rem, KL2Rem}.
+
+
+-spec split_unexpanded_files(
+    list(merge_maybe_expanded_pointer())) -> 
+        {
+            list(leveled_codec:ledger_kv()|leveled_sst:slot_pointer()),
+            list(leveled_sst:sst_pointer())
+        }. 
+split_unexpanded_files(Pointers) ->
+    split_unexpanded_files(Pointers, [], []).
+
+-spec split_unexpanded_files(
+    list(merge_maybe_expanded_pointer()),
+    list(leveled_codec:ledger_kv()|leveled_sst:slot_pointer()),
+    list(leveled_sst:sst_pointer())) -> 
+        {
+            list(leveled_codec:ledger_kv()|leveled_sst:slot_pointer()),
+            list(leveled_sst:sst_pointer())
+        }. 
+split_unexpanded_files([], MaybeExpanded, FilePointers) ->
+    {lists:reverse(MaybeExpanded), lists:reverse(FilePointers)};
+split_unexpanded_files([{next, P, SK}|Rest], MaybeExpanded, FilePointers) ->
+    split_unexpanded_files(Rest, MaybeExpanded, [{next, P, SK}|FilePointers]);
+split_unexpanded_files([{LK, LV}|Rest], MaybeExpanded, []) ->
+        % Should never see this, once a FilePointer has been seen
+    split_unexpanded_files(Rest, [{LK, LV}|MaybeExpanded], []);
+split_unexpanded_files([{pointer, P, SIV, SK, EK}|Rest], MaybeExpanded, []) ->
+        % Should never see this, once a FilePointer has been seen
+    split_unexpanded_files(
+        Rest, [{pointer, P, SIV, SK, EK}|MaybeExpanded], []
+    ).
 
 -spec grooming_scorer(
     list(leveled_pmanifest:manifest_entry()))