Merge branch 'master' into mas-i189-tuplebuckets

src/leveled_bookie.erl

@@ -48,6 +48,7 @@
         code_change/3,
         book_start/1,
         book_start/4,
+        book_plainstart/1,
         book_put/5,
         book_put/6,
         book_tempput/7,
@@ -357,6 +358,14 @@ book_start(Opts) ->
     gen_server:start_link(?MODULE, [set_defaults(Opts)], []).
 
 
+-spec book_plainstart(list(tuple())) -> {ok, pid()}.
+
+%% @doc
+%% Start used in tests to start without linking
+book_plainstart(Opts) ->
+    gen_server:start(?MODULE, [set_defaults(Opts)], []).
+
+
 -spec book_tempput(pid(), key(), key(), any(), 
                     leveled_codec:index_specs(), 
                     leveled_codec:tag(), integer()) -> ok|pause.

src/leveled_cdb.erl

@@ -625,11 +625,8 @@ reader({direct_fetch, PositionList, Info}, _From, State) ->
                         PositionList);
             key_value_check ->
                 BM = State#state.binary_mode,
-                lists:filtermap(
-                    fun(P) ->
-                            FilterFalseKey(extract_key_value_check(H, P, BM))
-                    end,
-                        PositionList)
+                lists:map(fun(P) -> extract_key_value_check(H, P, BM) end,
+                            PositionList)
         end,
     {reply, Reply, reader, State};
 reader(cdb_complete, _From, State) ->
@@ -1269,6 +1266,8 @@ scan_over_file(Handle, Position, FilterFun, Output, LastKey) ->
     case saferead_keyvalue(Handle) of
         false ->
             leveled_log:log("CDB09", [Position]),
+            % Bring file back to that position
+            {ok, Position} = file:position(Handle, {bof, Position}),
             {eof, Output};
         {Key, ValueAsBin, KeyLength, ValueLength} ->
             NewPosition = case Key of
@@ -1318,7 +1317,7 @@ saferead_keyvalue(Handle) ->
     case read_next_2_integers(Handle) of 
         eof ->
             false;
-        {KeyL, ValueL} ->
+        {KeyL, ValueL} when is_integer(KeyL), is_integer(ValueL) ->
             case safe_read_next_keybin(Handle, KeyL) of 
                 false ->
                     false;
@@ -1330,7 +1329,9 @@ saferead_keyvalue(Handle) ->
                             % i.e. value with no CRC
                             {Key, TrueValue, KeyL, ValueL}
                     end
-            end
+            end;
+        _ ->
+            false
     end.
 
 
@@ -1437,7 +1438,7 @@ extract_valueandsize(ValueAsBin) ->
 %% Note that the endian_flip is required to make the file format compatible 
 %% with CDB 
 read_next_2_integers(Handle) ->
-    case file:read(Handle,?DWORD_SIZE) of 
+    case file:read(Handle, ?DWORD_SIZE) of 
         {ok, <<Int1:32,Int2:32>>} -> 
             {endian_flip(Int1), endian_flip(Int2)};
         ReadError ->
@@ -2444,7 +2445,7 @@ crc_corrupt_writer_test() ->
     ok = file:close(Handle),
     {ok, P2} = cdb_open_writer("../test/corruptwrt_test.pnd",
                                 #cdb_options{binary_mode=false}),
-                                ?assertMatch(probably, cdb_keycheck(P2, "Key1")),
+    ?assertMatch(probably, cdb_keycheck(P2, "Key1")),
     ?assertMatch({"Key1", "Value1"}, cdb_get(P2, "Key1")),
     ?assertMatch(missing, cdb_get(P2, "Key100")),
     ok = cdb_put(P2, "Key100", "Value100"),
@@ -2532,7 +2533,57 @@ safe_read_test() ->
                     saferead_keyvalue(HandleHappy)),
     
     file:delete(TestFN).
+
+
+get_positions_corruption_test() ->
+    F1 = "../test/corruptpos_test.pnd",
+    file:delete(F1),
+    {ok, P1} = cdb_open_writer(F1, #cdb_options{binary_mode=false}),
+    KVList = generate_sequentialkeys(1000, []),
+    ok = cdb_mput(P1, KVList),
+    ?assertMatch(probably, cdb_keycheck(P1, "Key1")),
+    ?assertMatch({"Key1", "Value1"}, cdb_get(P1, "Key1")),
+    ?assertMatch({"Key100", "Value100"}, cdb_get(P1, "Key100")),
+    {ok, F2} = cdb_complete(P1),
+    {ok, P2} = cdb_open_reader(F2, #cdb_options{binary_mode=false}),
+    PositionList = cdb_getpositions(P2, all),
+    ?assertMatch(1000, length(PositionList)),
+    ok = cdb_close(P2),
+
+    {ok, Handle} = file:open(F2, ?WRITE_OPS),
+    Positions = lists:sublist(PositionList, 200, 10),
+    CorruptFun = 
+        fun(Offset) ->
+            ok = file:pwrite(Handle, Offset, <<0:8/integer>>)
+        end,
+    ok = lists:foreach(CorruptFun, Positions),
+    ok = file:close(Handle),
     
+    {ok, P3} = cdb_open_reader(F2, #cdb_options{binary_mode=false}),
+    
+    PositionList = cdb_getpositions(P3, all),
+    ?assertMatch(1000, length(PositionList)),
+    
+    KVCL = cdb_directfetch(P3, PositionList, key_size),
+    ?assertMatch(true, length(KVCL) < 1000),
+    ok = cdb_close(P3),
+    file:delete(F2).
+    
+badly_written_test() ->
+    F1 = "../test/badfirstwrite_test.pnd",
+    file:delete(F1),
+    {ok, Handle} = file:open(F1, ?WRITE_OPS),
+    ok = file:pwrite(Handle, 256 * ?DWORD_SIZE, <<1:8/integer>>),
+    ok = file:close(Handle),
+    {ok, P1} = cdb_open_writer(F1, #cdb_options{binary_mode=false}),
+    ok = cdb_put(P1, "Key100", "Value100"),
+    ?assertMatch({"Key100", "Value100"}, cdb_get(P1, "Key100")),
+    ok = cdb_close(P1),
+    {ok, P2} = cdb_open_writer(F1, #cdb_options{binary_mode=false}),
+    ?assertMatch({"Key100", "Value100"}, cdb_get(P2, "Key100")),
+    ok = cdb_close(P2),
+    file:delete(F1).
+
 
 nonsense_coverage_test() ->
     {ok, Pid} = gen_fsm:start_link(?MODULE, [#cdb_options{}], []),

src/leveled_log.erl

@@ -364,7 +364,7 @@
                     ++ "with totals of cycle_count=~w "
                     ++ "fetch_time=~w index_time=~w"}},
     {"CDB20",
-        {error, "Error ~w caught when safe reading a file to length ~w"}}
+        {warn, "Error ~w caught when safe reading a file to length ~w"}}
         ]).