add segment copy op

Author: kjnilsson

docs/internals/COMPACTION.md

@@ -101,6 +101,8 @@ e.g. the compacted segment `001.compacting` to `001.segment` without breaking
 any references to the segment. Single segment compaction should only be triggered
 when a certain limit has been reached, e.g. > 50% of indexes can be cleared up.
 
+TODO: how to handle compaction of segments that have indexes that never were
+committed, i.e. overwritten?
 
 
 #### When does phase 3 compaction run?

src/ra_log.erl

@@ -223,6 +223,8 @@ init(#{uid := UId,
     % this queries the segment writer and thus blocks until any
     % segments it is currently processed have been finished
     MtRange = ra_mt:range(Mt0),
+    %% TODO: init ra_log_reader here instead and let it take care of range
+    %% calulation and segref compaction
     {{FirstIdx, LastIdx0}, SegRefs} = case recover_ranges(UId, MtRange, SegWriter) of
                                           {undefined, SRs} ->
                                               {{-1, -1}, SRs};
@@ -938,31 +940,30 @@ install_snapshot({SnapIdx, SnapTerm} = IdxTerm, MacMod, LiveIndexes,
     put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_INDEX, SnapIdx),
     put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_WRITTEN_INDEX, SnapIdx),
 
-    CompEffs = schedule_compaction(SnapIdx, State0),
-
     {SnapState, Checkpoints} =
         ra_snapshot:take_older_checkpoints(SnapIdx, SnapState0),
     CPEffects = [{delete_snapshot,
                   ra_snapshot:directory(SnapState, checkpoint),
                   Checkpoint} || Checkpoint <- Checkpoints],
-    State = State0#?MODULE{snapshot_state = SnapState,
-                           first_index = SnapIdx + 1,
-                           last_index = SnapIdx,
-                           last_term = SnapTerm,
-                           last_written_index_term = IdxTerm},
-    %% TODO: more mt entries could potentially be cleared up in the
-    %% mem table here
     SmallestLiveIndex = case ra_seq:first(LiveIndexes) of
                             undefined ->
                                 SnapIdx + 1;
                             I ->
                                 I
                         end,
+    %% TODO: more mt entries could potentially be cleared up in the
+    %% mem table here
     {Spec, Mt} = ra_mt:set_first(SmallestLiveIndex, Mt0),
     ok = exec_mem_table_delete(Names, UId, Spec),
-    {ok, State#?MODULE{live_indexes = LiveIndexes,
-                       mem_table = Mt},
-     CompEffs ++ CPEffects}.
+    State = State0#?MODULE{snapshot_state = SnapState,
+                           first_index = SnapIdx + 1,
+                           last_index = SnapIdx,
+                           last_term = SnapTerm,
+                           live_indexes = LiveIndexes,
+                           mem_table = Mt,
+                           last_written_index_term = IdxTerm},
+    CompEffs = schedule_compaction(SnapIdx, State),
+    {ok, State, CompEffs ++ CPEffects}.
 
 
 -spec recover_snapshot(State :: state()) ->
@@ -1261,6 +1262,9 @@ schedule_compaction(SnapIdx, #?MODULE{cfg = #cfg{},
                                       end, lists:reverse(Compactable)),
             SnapDir = ra_snapshot:current_snapshot_dir(SnapState),
 
+            %% TODO: minor compactions should also delete / truncate
+            %% segments with completely overwritten indexes
+
             Self = self(),
             Fun = fun () ->
                           {ok, Indexes} = ra_snapshot:indexes(SnapDir),

src/ra_log_reader.erl

@@ -109,6 +109,7 @@ update_segments(NewSegmentRefs,
                         segment_refs = SegRefs0} = State) ->
 
     SegmentRefs0 = ra_lol:to_list(SegRefs0),
+    %% TODO: capture segrefs removed by compact_segrefs/2 and delete them
     SegmentRefsComp = compact_segrefs(NewSegmentRefs, SegmentRefs0),
     SegmentRefsCompRev = lists:reverse(SegmentRefsComp),
     SegRefs = ra_lol:from_list(fun seg_ref_gt/2, SegmentRefsCompRev),
@@ -470,13 +471,13 @@ compact_seg_refs_test() ->
 
 compact_segref_3_test() ->
     Data = [
-            {{2, 7}, "B"},
+            {{2, 7}, "C"},
             %% this entry has overwritten the prior two
             {{5, 10}, "B"},
             {{1, 4}, "A"}
            ],
     Res = compact_segrefs(Data, []),
-    ?assertMatch([{{2, 7}, "B"},
+    ?assertMatch([{{2, 7}, "C"},
                   {{1, 1}, "A"}], Res),
     ok.
 

src/ra_log_segment.erl

@@ -22,13 +22,15 @@
          max_count/1,
          filename/1,
          segref/1,
-         is_same_as/2]).
+         is_same_as/2,
+         copy/3]).
 
 -export([dump/1,
          dump_index/1]).
 
 -include("ra.hrl").
 
+-include_lib("stdlib/include/assert.hrl").
 -include_lib("kernel/include/file.hrl").
 
 -define(VERSION, 2).
@@ -474,6 +476,23 @@ close(#state{cfg = #cfg{fd = Fd}}) ->
     _ = file:close(Fd),
     ok.
 
+-spec copy(state(), file:filename_all(), [ra:index()]) ->
+    {ok, state()} | {error, term()}.
+copy(#state{} = State0, FromFile, Indexes)
+  when is_list(Indexes) ->
+    {ok, From} = open(FromFile, #{mode => read}),
+    %% TODO: the current approach recalculates the CRC and isn't completely
+    %% optimial. Also it does not allow for a future where copy_file_range may
+    %% be available
+    State = lists:foldl(
+              fun (I, S0) ->
+                      {ok, Term, Data} = simple_read(From, I),
+                      {ok, S} = append(S0, I, Term, Data),
+                      S
+              end, State0, lists:sort(Indexes)),
+    close(From),
+    sync(State).
+
 %%% Internal
 
 is_same_filename_all(Fn, Fn) ->
@@ -693,6 +712,20 @@ is_full(#state{cfg = #cfg{max_size = MaxSize},
     IndexOffset >= DataStart orelse
     (DataOffset - DataStart) > MaxSize.
 
+simple_read(#state{cfg = #cfg{fd = Fd},
+                   index = SegIndex}, Idx)
+  when is_map_key(Idx, SegIndex) ->
+    {Term, Pos, Len, _} = map_get(Idx, SegIndex),
+    case file:pread(Fd, Pos, Len) of
+        {ok, Data} ->
+            ?assert(byte_size(Data) == Len),
+            {ok, Term, Data};
+        Err ->
+            Err
+    end;
+simple_read(_State, _) ->
+    {error, not_found}.
+
 -ifdef(TEST).
 -include_lib("eunit/include/eunit.hrl").
 

test/ra_log_segment_SUITE.erl

@@ -38,7 +38,8 @@ all_tests() ->
      corrupted_segment,
      large_segment,
      segref,
-     versions_v1
+     versions_v1,
+     copy
     ].
 
 groups() ->
@@ -460,16 +461,44 @@ read_sparse_append_read(Config) ->
     ra_log_segment:close(R0),
     ok.
 
+copy(Config) ->
+    Dir = ?config(data_dir, Config),
+    Indexes = lists:seq(1, 100),
+    SrcFn = filename:join(Dir, <<"SOURCE1.segment">>),
+    {ok, SrcSeg0} = ra_log_segment:open(SrcFn),
+    SrcSeg1 = lists:foldl(
+             fun (I, S0) ->
+                     {ok, S} = ra_log_segment:append(S0, I, 1, term_to_binary(I)),
+                     S
+             end, SrcSeg0, Indexes),
+    _ = ra_log_segment:close(SrcSeg1),
+
+    Fn = filename:join(Dir, <<"TARGET.segment">>),
+    {ok, Seg0} = ra_log_segment:open(Fn),
+    CopyIndexes = lists:seq(1, 100, 2),
+    {ok, Seg} = ra_log_segment:copy(Seg0, SrcFn, CopyIndexes),
+    ra_log_segment:close(Seg),
+    {ok, R} = ra_log_segment:open(Fn, #{mode => read,
+                                        access_pattern => random}),
+    %%TODO: consider makeing read_sparse tolerant to missing indexes somehow
+    %% perhaps detecting if the segment is "sparse"
+    {ok, 2, [_, _]} = ra_log_segment:read_sparse(R, [1, 3],
+                                                 fun (I, T, B, Acc) ->
+                                                         [{I, T, binary_to_term(B)} | Acc]
+                                                 end, []),
+    ra_log_segment:close(R),
+
+    ok.
+
+
+%%% Internal
 write_until_full(Idx, Term, Data, Seg0) ->
     case ra_log_segment:append(Seg0, Idx, Term, Data) of
         {ok, Seg} ->
             write_until_full(Idx+1, Term, Data, Seg);
         {error, full} ->
             Seg0
     end.
-
-
-%%% Internal
 make_data(Size) ->
     term_to_binary(crypto:strong_rand_bytes(Size)).