only flush live indexes in seg writer

Author: kjnilsson

src/ra_kv.erl

@@ -21,7 +21,8 @@
 
          put/4,
          get/3,
-         query_get/3
+         query_get/3,
+         take_snapshot/1
         ]).
 
 
@@ -47,7 +48,9 @@
               command/0]).
 
 %% mgmt
--spec start_cluster(atom(), atom(), map()) ->
+-spec start_cluster(System :: atom(),
+                    ClusterName :: atom(),
+                    Config :: #{members := [ra_server_id()]}) ->
     {ok, [ra_server_id()], [ra_server_id()]} |
     {error, cluster_not_formed}.
 start_cluster(System, ClusterName, #{members := ServerIds})
@@ -134,6 +137,10 @@ query_get(ClusterName, Key, #?STATE{keys = Keys}) ->
             {error, not_found}
     end.
 
+-spec take_snapshot(ra_server_id()) -> ok.
+take_snapshot(ServerId) ->
+    ra:aux_command(ServerId, take_snapshot).
+
 %% state machine
 
 init(_) ->
@@ -154,6 +161,7 @@ live_indexes(#?STATE{keys = Keys}) ->
                       [Idx | Acc]
               end, [], Keys).
 
+
 -record(aux, {}).
 
 init_aux(_) ->

src/ra_log_segment_writer.erl

@@ -264,28 +264,41 @@ get_overview(#state{data_dir = Dir,
 flush_mem_table_ranges({ServerUId, TidSeqs0},
                        #state{system = System} = State) ->
     SmallestIdx = smallest_live_idx(ServerUId),
-    %% TidRanges arrive here sorted new -> old.
+    LiveIndexes = live_indexes(ServerUId),
+    LastLive = ra_seq:last(LiveIndexes),
+    %% TidSeqs arrive here sorted new -> old.
 
-    %% truncate and limit all ranges to create a contiguous non-overlapping
+    %% TODO: use live indexes from ra_log_snapshot_state table to only
+    %% write live entries below the snapshot index
+
+    %% truncate and limit all seqa to create a contiguous non-overlapping
     %% list of tid ranges to flush to disk
-    %% now TidRanges are sorted old -> new, i.e the correct order of
-    %% processing
     TidSeqs = lists:foldl(
                 fun ({T, Seq0}, []) ->
                         case ra_seq:floor(SmallestIdx, Seq0) of
                             [] ->
                                 [];
+                            Seq when LiveIndexes == []->
+                                [{T, Seq}];
                             Seq ->
-                                [{T, Seq}]
+                                L = ra_seq:in_range(ra_seq:range(Seq),
+                                                    LiveIndexes),
+
+                                [{T, ra_seq:add(ra_seq:floor(LastLive + 1, Seq), L)}]
                         end;
                     ({T, Seq0}, [{_T, PrevSeq} | _] = Acc) ->
                         Start = ra_seq:first(PrevSeq),
                         Seq1 = ra_seq:floor(SmallestIdx, Seq0),
                         case ra_seq:limit(Start, Seq1) of
                             [] ->
                                 Acc;
+                            Seq when LiveIndexes == [] ->
+                                [{T, Seq} | Acc];
                             Seq ->
-                                [{T, Seq} | Acc]
+                                L = ra_seq:in_range(ra_seq:range(Seq),
+                                                    LiveIndexes),
+                                [{T, ra_seq:add(ra_seq:floor(LastLive + 1, Seq), L)}
+                                | Acc]
                         end
                 end, [], TidSeqs0),
 
@@ -357,6 +370,9 @@ start_index(ServerUId, StartIdx0) ->
 smallest_live_idx(ServerUId) ->
     ra_log_snapshot_state:smallest(ra_log_snapshot_state, ServerUId).
 
+live_indexes(ServerUId) ->
+    ra_log_snapshot_state:live_indexes(ra_log_snapshot_state, ServerUId).
+
 send_segments(System, ServerUId, TidRanges, SegRefs) ->
     case ra_directory:pid_of(System, ServerUId) of
         undefined ->

src/ra_log_snapshot_state.erl

@@ -4,6 +4,7 @@
          insert/5,
          delete/2,
          smallest/2,
+         live_indexes/2,
          snapshot/2
         ]).
 
@@ -26,6 +27,11 @@ delete(Table, UId) ->
 smallest(Table, UId) when is_binary(UId) ->
     ets:lookup_element(Table, UId, 3, 0).
 
+-spec live_indexes(ets:table(), ra:uid()) ->
+    ra:index().
+live_indexes(Table, UId) when is_binary(UId) ->
+    ets:lookup_element(Table, UId, 4, []).
+
 -spec snapshot(ets:table(), ra:uid()) ->
     ra:index() | -1.
 snapshot(Table, UId) when is_binary(UId) ->

src/ra_seq.erl

@@ -89,15 +89,19 @@ limit(CeilIdx, [{_, _} = T | Rem]) ->
 limit(_CeilIdx, Seq) ->
     Seq.
 
--spec add(state(), state()) -> state().
-add([], Seq2) ->
-    Seq2;
-add(Seq1, Seq2) ->
-    Fst = case lists:last(Seq1) of
-              {I, _} -> I;
-              I -> I
-          end,
-    fold(fun append/2, limit(Fst - 1, Seq2), Seq1).
+%% @doc adds two sequences together where To is
+%% the "lower" sequence
+-spec add(Add :: state(), To :: state()) -> state().
+add([], To) ->
+    To;
+add(Add, []) ->
+    Add;
+add(Add, To) ->
+    Fst = first(Add),
+              % {I, _} -> I;
+              % I -> I
+          % end,
+    fold(fun append/2, limit(Fst - 1, To), Add).
 
 -spec fold(fun ((ra:index(), Acc) -> Acc), Acc, state()) ->
     Acc when Acc :: term().
@@ -199,6 +203,8 @@ range(Seq) ->
     ra_range:new(first(Seq), last(Seq)).
 
 
+-spec in_range(ra:range(), state()) ->
+    state().
 in_range(_Range, []) ->
     [];
 in_range(undefined, _) ->

test/ra_log_segment_writer_SUITE.erl

@@ -41,7 +41,8 @@ all_tests() ->
      my_segments,
      upgrade_segment_name_format,
      skip_entries_lower_than_snapshot_index,
-     skip_all_entries_lower_than_snapshot_index
+     skip_all_entries_lower_than_snapshot_index,
+     live_indexes_1
     ].
 
 groups() ->
@@ -839,6 +840,47 @@ skip_all_entries_lower_than_snapshot_index(Config) ->
     ok = gen_server:stop(TblWriterPid),
     ok.
 
+live_indexes_1(Config) ->
+    Dir = ?config(wal_dir, Config),
+    UId = ?config(uid, Config),
+    {ok, TblWriterPid} = ra_log_segment_writer:start_link(#{system => default,
+                                                            name => ?SEGWR,
+                                                            data_dir => Dir}),
+    % first batch
+    Entries = [{1, 42, a},
+               {2, 42, b},
+               {3, 43, c},
+               {4, 43, d},
+               {5, 43, e},
+               {6, 43, f}
+              ],
+    Mt = make_mem_table(UId, Entries),
+    Ranges = #{UId => [{ra_mt:tid(Mt), [ra_mt:range(Mt)]}]},
+    %% update snapshot state table
+    ra_log_snapshot_state:insert(ra_log_snapshot_state, UId, 4, 2, [4, 2]),
+    ok = ra_log_segment_writer:accept_mem_tables(?SEGWR, Ranges,
+                                                 make_wal(Config, "w1.wal")),
+    receive
+        {ra_log_event, {segments, _Tid, [{Fn, {2, 6}}]}} ->
+            SegmentFile = filename:join(?config(server_dir, Config), Fn),
+            {ok, Seg} = ra_log_segment:open(SegmentFile, #{mode => read}),
+            % assert only entries with a higher index than the snapshot
+            % have been written
+            ok = gen_server:stop(TblWriterPid),
+            ?assertExit({missing_key, 3}, read_sparse(Seg, [2, 3, 4])),
+            [
+             {2, _, _},
+             {4, _, _},
+             {5, _, _},
+             {6, _, _}
+            ] = read_sparse(Seg, [2, 4, 5, 6])
+    after 3000 ->
+              flush(),
+              ok = gen_server:stop(TblWriterPid),
+              throw(ra_log_event_timeout)
+    end,
+    ok.
+
 %%% Internal
 
 fake_mem_table(UId, Dir, Entries) ->