umgeher/leveled
0
Fork 0
Code Issues Pull requests Releases Activity Actions

Merge pull request #139 from martinsumner/mas-i138-useblockindexonfetchrange

Browse source 
Mas i138 useblockindexonfetchrange
This commit is contained in:
Martin Sumner 2018-05-17 19:32:31 +01:00 • committed by GitHub
commit 501b7806e9
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 280 additions and 164 deletions
Download patch file Download diff file Expand all files Collapse all files

258
src/leveled_sst.erl
View file

@ -145,7 +145,8 @@
:: {next, :: {next,
leveled_pmanifest:manifest_entry(), leveled_pmanifest:manifest_entry(),
leveled_codec:ledger_key()|all}. leveled_codec:ledger_key()|all}.
-type binaryslot_element()
:: {tuple(), tuple()}|{binary(), integer(), tuple(), tuple()}.
%% yield_blockquery is used to detemrine if the work necessary to process a %% yield_blockquery is used to detemrine if the work necessary to process a
%% range query beyond the fetching the slot should be managed from within %% range query beyond the fetching the slot should be managed from within
@ -375,8 +376,9 @@ sst_getfilteredrange(Pid, StartKey, EndKey, ScanWidth, SegList) ->
ScanWidth, SegList0}, ScanWidth, SegList0},
infinity) of infinity) of
{yield, SlotsToFetchBinList, SlotsToPoint, PressMethod} -> {yield, SlotsToFetchBinList, SlotsToPoint, PressMethod} ->
binaryslot_reader(SlotsToFetchBinList, PressMethod) {L, _BIC} =
++ SlotsToPoint; binaryslot_reader(SlotsToFetchBinList, PressMethod, SegList0),
L ++ SlotsToPoint;
Reply -> Reply ->
Reply Reply
end. end.
@ -405,7 +407,8 @@ sst_getfilteredslots(Pid, SlotList, SegList) ->
SegL0 = tune_seglist(SegList), SegL0 = tune_seglist(SegList),
{SlotBins, PressMethod} = {SlotBins, PressMethod} =
gen_fsm:sync_send_event(Pid, {get_slots, SlotList, SegL0}, infinity), gen_fsm:sync_send_event(Pid, {get_slots, SlotList, SegL0}, infinity),
binaryslot_reader(SlotBins, PressMethod). {L, _BIC} = binaryslot_reader(SlotBins, PressMethod, SegL0),
L.
-spec sst_getmaxsequencenumber(pid()) -> integer(). -spec sst_getmaxsequencenumber(pid()) -> integer().
%% @doc %% @doc
@ -569,13 +572,15 @@ reader({get_kv, LedgerKey, Hash}, _From, State) ->
{reply, Result, reader, UpdState#state{timings = UpdTimings0, {reply, Result, reader, UpdState#state{timings = UpdTimings0,
timings_countdown = CountDown}}; timings_countdown = CountDown}};
reader({get_kvrange, StartKey, EndKey, ScanWidth, SlotList}, _From, State) -> reader({get_kvrange, StartKey, EndKey, ScanWidth, SegList}, _From, State) ->
{SlotsToFetchBinList, SlotsToPoint} = fetch_range(StartKey, {SlotsToFetchBinList, SlotsToPoint} = fetch_range(StartKey,
EndKey, EndKey,
ScanWidth, ScanWidth,
SlotList, SegList,
State), State),
PressMethod = State#state.compression_method, PressMethod = State#state.compression_method,
case State#state.yield_blockquery of case State#state.yield_blockquery of
true -> true ->
{reply, {reply,
@ -586,11 +591,22 @@ reader({get_kvrange, StartKey, EndKey, ScanWidth, SlotList}, _From, State) ->
reader, reader,
State}; State};
false -> false ->
{L, BIC} =
binaryslot_reader(SlotsToFetchBinList, PressMethod, SegList),
FoldFun =
fun(CacheEntry, Cache) ->
case CacheEntry of
{_ID, none} ->
Cache;
{ID, Header} ->
array:set(ID - 1, Header, Cache)
end
end,
BlockIdxC0 = lists:foldl(FoldFun, State#state.blockindex_cache, BIC),
{reply, {reply,
binaryslot_reader(SlotsToFetchBinList, PressMethod) L ++ SlotsToPoint,
++ SlotsToPoint,
reader, reader,
State} State#state{blockindex_cache = BlockIdxC0}}
end; end;
reader({get_slots, SlotList, SegList}, _From, State) -> reader({get_slots, SlotList, SegList}, _From, State) ->
SlotBins = SlotBins =
@ -629,12 +645,12 @@ reader(close, _From, State) ->
delete_pending({get_kv, LedgerKey, Hash}, _From, State) -> delete_pending({get_kv, LedgerKey, Hash}, _From, State) ->
{Result, UpdState, _Ts} = fetch(LedgerKey, Hash, State, no_timing), {Result, UpdState, _Ts} = fetch(LedgerKey, Hash, State, no_timing),
{reply, Result, delete_pending, UpdState, ?DELETE_TIMEOUT}; {reply, Result, delete_pending, UpdState, ?DELETE_TIMEOUT};
delete_pending({get_kvrange, StartKey, EndKey, ScanWidth, SlotList}, delete_pending({get_kvrange, StartKey, EndKey, ScanWidth, SegList},
_From, State) -> _From, State) ->
{SlotsToFetchBinList, SlotsToPoint} = fetch_range(StartKey, {SlotsToFetchBinList, SlotsToPoint} = fetch_range(StartKey,
EndKey, EndKey,
ScanWidth, ScanWidth,
SlotList, SegList,
State), State),
% Always yield as about to clear and de-reference % Always yield as about to clear and de-reference
PressMethod = State#state.compression_method, PressMethod = State#state.compression_method,
@ -756,8 +772,7 @@ fetch(LedgerKey, Hash, State, Timings0) ->
StartPos = Slot#slot_index_value.start_position, StartPos = Slot#slot_index_value.start_position,
Result = Result =
check_blocks(PosList, check_blocks(PosList,
State#state.handle, {State#state.handle, StartPos},
StartPos,
BlockLengths, BlockLengths,
byte_size(PosBin), byte_size(PosBin),
LedgerKey, LedgerKey,
@ -778,8 +793,8 @@ fetch(LedgerKey, Hash, State, Timings0) ->
end. end.
-spec fetch_range(tuple(), tuple(), integer(), false|list(integer()), sst_state()) -spec fetch_range(tuple(), tuple(), integer(), leveled_codec:segment_list(),
-> {list(), list()}. sst_state()) -> {list(), list()}.
%% @doc %% @doc
%% Fetch the contents of the SST file for a given key range. This will %% Fetch the contents of the SST file for a given key range. This will
%% pre-fetch some results, and append pointers for additional results. %% pre-fetch some results, and append pointers for additional results.
@ -1254,23 +1269,30 @@ generate_binary_slot(Lookup, KVL, PressMethod, BuildTimings0) ->
{{Header, SlotBin, HashL, LastKey}, BuildTimings3}. {{Header, SlotBin, HashL, LastKey}, BuildTimings3}.
% Acc should start as not_present if LedgerKey is a key, and a list if
% LedgerKey is false
check_blocks([], _Handle, _StartPos, _BlockLengths, _PosBinLength,
-spec check_blocks(list(integer()),
binary()|{file:io_device(), integer()},
binary(),
integer(),
leveled_codec:ledger_key()|false,
press_methods(),
list()|not_present) -> list()|not_present.
%% @doc
%% Acc should start as not_present if LedgerKey is a key, and a list if
%% LedgerKey is false
check_blocks([], _BlockPointer, _BlockLengths, _PosBinLength,
_LedgerKeyToCheck, _PressMethod, not_present) ->
not_present;
check_blocks([], _BlockPointer, _BlockLengths, _PosBinLength,
_LedgerKeyToCheck, _PressMethod, Acc) -> _LedgerKeyToCheck, _PressMethod, Acc) ->
case is_list(Acc) of
true ->
lists:reverse(Acc); lists:reverse(Acc);
false -> check_blocks([Pos|Rest], BlockPointer, BlockLengths, PosBinLength,
Acc
end;
check_blocks([Pos|Rest], Handle, StartPos, BlockLengths, PosBinLength,
LedgerKeyToCheck, PressMethod, Acc) -> LedgerKeyToCheck, PressMethod, Acc) ->
{BlockNumber, BlockPos} = revert_position(Pos), {BlockNumber, BlockPos} = revert_position(Pos),
BlockBin = BlockBin =
read_block(Handle, read_block(BlockPointer,
StartPos,
BlockLengths, BlockLengths,
PosBinLength, PosBinLength,
BlockNumber), BlockNumber),
@ -1282,19 +1304,19 @@ check_blocks([Pos|Rest], Handle, StartPos, BlockLengths, PosBinLength,
_ -> _ ->
case LedgerKeyToCheck of case LedgerKeyToCheck of
false -> false ->
check_blocks(Rest, Handle, StartPos, check_blocks(Rest, BlockPointer,
BlockLengths, PosBinLength, BlockLengths, PosBinLength,
LedgerKeyToCheck, PressMethod, LedgerKeyToCheck, PressMethod,
[{K, V}|Acc]); [{K, V}|Acc]);
_ -> _ ->
check_blocks(Rest, Handle, StartPos, check_blocks(Rest, BlockPointer,
BlockLengths, PosBinLength, BlockLengths, PosBinLength,
LedgerKeyToCheck, PressMethod, Acc) LedgerKeyToCheck, PressMethod, Acc)
end end
end. end.
read_block(Handle, StartPos, BlockLengths, PosBinLength, BlockID) -> read_block({Handle, StartPos}, BlockLengths, PosBinLength, BlockID) ->
{Offset, Length} = block_offsetandlength(BlockLengths, BlockID), {Offset, Length} = block_offsetandlength(BlockLengths, BlockID),
{ok, BlockBin} = file:pread(Handle, {ok, BlockBin} = file:pread(Handle,
StartPos StartPos
@ -1304,6 +1326,11 @@ read_block(Handle, StartPos, BlockLengths, PosBinLength, BlockID) ->
% 4-byte CRC, 4-byte pos, % 4-byte CRC, 4-byte pos,
% 4-byte CRC, 5x4 byte lengths % 4-byte CRC, 5x4 byte lengths
Length), Length),
BlockBin;
read_block(SlotBin, BlockLengths, PosBinLength, BlockID) ->
{Offset, Length} = block_offsetandlength(BlockLengths, BlockID),
StartPos = Offset + PosBinLength + 32,
<<_Pre:StartPos/binary, BlockBin:Length/binary, _Rest/binary>> = SlotBin,
BlockBin. BlockBin.
read_slot(Handle, Slot) -> read_slot(Handle, Slot) ->
@ -1333,19 +1360,20 @@ pointer_mapfun(Pointer) ->
%% Return a function that can pull individual slot binaries from a binary %% Return a function that can pull individual slot binaries from a binary
%% covering multiple slots %% covering multiple slots
binarysplit_mapfun(MultiSlotBin, StartPos) -> binarysplit_mapfun(MultiSlotBin, StartPos) ->
fun({SP, L, _ID, SK, EK}) -> fun({SP, L, ID, SK, EK}) ->
Start = SP - StartPos, Start = SP - StartPos,
<<_Pre:Start/binary, SlotBin:L/binary, _Post/binary>> = MultiSlotBin, <<_Pre:Start/binary, SlotBin:L/binary, _Post/binary>> = MultiSlotBin,
{SlotBin, SK, EK} {SlotBin, ID, SK, EK}
end. end.
-spec read_slots(file:io_device(), list(), -spec read_slots(file:io_device(), list(),
{false|list(), any()}, press_methods()) -> list(). {false|list(), any()}, press_methods())
-> list(binaryslot_element()).
%% @doc %% @doc
%% The reading of sots will return a list of either 2-tuples containing %% The reading of sots will return a list of either 2-tuples containing
%% {K, V} pairs - or 3-tuples containing {Binary, SK, EK}. The 3 tuples %% {K, V} pairs - or 3-tuples containing {Binary, SK, EK}. The 3 tuples
%% can be exploded into lists of {K, V} pairs using the binaryslot_reader/2 %% can be exploded into lists of {K, V} pairs using the binaryslot_reader/3
%% function %% function
%% %%
%% Reading slots is generally unfiltered, but in the sepcial case when %% Reading slots is generally unfiltered, but in the sepcial case when
@ -1392,7 +1420,7 @@ read_slots(Handle, SlotList, {SegList, BlockIndexCache}, PressMethod) ->
PositionList -> PositionList ->
Acc ++ Acc ++
check_blocks(PositionList, check_blocks(PositionList,
Handle, SP, {Handle, SP},
BlockLengths, BlockLengths,
byte_size(BlockIdx), byte_size(BlockIdx),
false, PressMethod, false, PressMethod,
@ -1403,24 +1431,33 @@ read_slots(Handle, SlotList, {SegList, BlockIndexCache}, PressMethod) ->
lists:foldl(BinMapFun, [], SlotList). lists:foldl(BinMapFun, [], SlotList).
-spec binaryslot_reader(list({tuple(), tuple()}|{binary(), tuple(), tuple()}), -spec binaryslot_reader(list(binaryslot_element()),
native|lz4) -> list({tuple(), tuple()}). native|lz4,
leveled_codec:segment_list())
-> {list({tuple(), tuple()}),
list({integer(), binary()})}.
%% @doc %% @doc
%% Read the binary slots converting them to {K, V} pairs if they were not %% Read the binary slots converting them to {K, V} pairs if they were not
%% already {K, V} pairs %% already {K, V} pairs
binaryslot_reader(SlotBinsToFetch, PressMethod) -> binaryslot_reader(SlotBinsToFetch, PressMethod, SegList) ->
binaryslot_reader(SlotBinsToFetch, PressMethod, []). binaryslot_reader(SlotBinsToFetch, PressMethod, SegList, [], []).
binaryslot_reader([], _PressMethod, Acc) -> binaryslot_reader([], _PressMethod, _SegList, Acc, BIAcc) ->
Acc; {Acc, BIAcc};
binaryslot_reader([{SlotBin, SK, EK}|Tail], PressMethod, Acc) -> binaryslot_reader([{SlotBin, ID, SK, EK}|Tail],
PressMethod, SegList, Acc, BIAcc) ->
{TrimmedL, BICache} =
binaryslot_trimmedlist(SlotBin,
SK, EK,
PressMethod,
SegList),
binaryslot_reader(Tail, binaryslot_reader(Tail,
PressMethod, PressMethod,
Acc ++ binaryslot_trimmedlist(SlotBin, SegList,
SK, EK, Acc ++ TrimmedL,
PressMethod)); [{ID, BICache}|BIAcc]);
binaryslot_reader([{K, V}|Tail], PressMethod, Acc) -> binaryslot_reader([{K, V}|Tail], PressMethod, SegList, Acc, BIAcc) ->
binaryslot_reader(Tail, PressMethod, Acc ++ [{K, V}]). binaryslot_reader(Tail, PressMethod, SegList, Acc ++ [{K, V}], BIAcc).
read_length_list(Handle, LengthList) -> read_length_list(Handle, LengthList) ->
@ -1478,12 +1515,48 @@ binaryslot_tolist(FullBin, PressMethod) ->
Out. Out.
binaryslot_trimmedlist(FullBin, all, all, PressMethod) -> binaryslot_trimmedlist(FullBin, all, all, PressMethod, false) ->
binaryslot_tolist(FullBin, PressMethod); {binaryslot_tolist(FullBin, PressMethod), none};
binaryslot_trimmedlist(FullBin, StartKey, EndKey, PressMethod) -> binaryslot_trimmedlist(FullBin, StartKey, EndKey, PressMethod, SegList) ->
LTrimFun = fun({K, _V}) -> K < StartKey end, LTrimFun = fun({K, _V}) -> K < StartKey end,
RTrimFun = fun({K, _V}) -> not leveled_codec:endkey_passed(EndKey, K) end, RTrimFun = fun({K, _V}) -> not leveled_codec:endkey_passed(EndKey, K) end,
BlockCheckFun =
fun(Block, {Acc, Continue}) ->
case {Block, Continue} of
{<<>>, _} ->
{Acc, false};
{_, true} ->
BlockList =
case is_binary(Block) of
true ->
deserialise_block(Block, PressMethod);
false ->
Block
end,
case fetchend_rawblock(BlockList) of
{LastKey, _LV} when StartKey > LastKey ->
{Acc, true};
{LastKey, _LV} ->
{_LDrop, RKeep} = lists:splitwith(LTrimFun,
BlockList),
case leveled_codec:endkey_passed(EndKey,
LastKey) of
true ->
{LKeep, _RDrop}
= lists:splitwith(RTrimFun, RKeep),
{Acc ++ LKeep, false};
false ->
{Acc ++ RKeep, true}
end;
_ ->
{Acc, true}
end;
{_ , false} ->
{Acc, false}
end
end,
case {crc_check_slot(FullBin), SegList} of
% It will be more effecient to check a subset of blocks. To work out % It will be more effecient to check a subset of blocks. To work out
% the best subset we always look in the middle block of 5, and based on % the best subset we always look in the middle block of 5, and based on
% the first and last keys of that middle block when compared to the Start % the first and last keys of that middle block when compared to the Start
@ -1492,9 +1565,7 @@ binaryslot_trimmedlist(FullBin, StartKey, EndKey, PressMethod) ->
% This isn't perfectly efficient, esepcially if the query overlaps Block2 % This isn't perfectly efficient, esepcially if the query overlaps Block2
% and Block3 (as Block 1 will also be checked), but finessing this last % and Block3 (as Block 1 will also be checked), but finessing this last
% scenario is hard to do in concise code % scenario is hard to do in concise code
BlocksToCheck = {{Header, Blocks}, false} ->
case crc_check_slot(FullBin) of
{Header, Blocks} ->
<<B1L:32/integer, <<B1L:32/integer,
B2L:32/integer, B2L:32/integer,
B3L:32/integer, B3L:32/integer,
@ -1504,6 +1575,7 @@ binaryslot_trimmedlist(FullBin, StartKey, EndKey, PressMethod) ->
<<Block1:B1L/binary, Block2:B2L/binary, <<Block1:B1L/binary, Block2:B2L/binary,
MidBlock:B3L/binary, MidBlock:B3L/binary,
Block4:B4L/binary, Block5:B5L/binary>> = Blocks, Block4:B4L/binary, Block5:B5L/binary>> = Blocks,
BlocksToCheck =
case B3L of case B3L of
0 -> 0 ->
[Block1, Block2]; [Block1, Block2];
@ -1532,48 +1604,25 @@ binaryslot_trimmedlist(FullBin, StartKey, EndKey, PressMethod) ->
_ -> _ ->
[Block1, Block2, MidBlockList, Block4, Block5] [Block1, Block2, MidBlockList, Block4, Block5]
end end
end;
crc_wonky ->
[]
end, end,
BlockCheckFun =
fun(Block, {Acc, Continue}) ->
case {Block, Continue} of
{<<>>, _} ->
{Acc, false};
{_, true} ->
BlockList =
case is_binary(Block) of
true ->
deserialise_block(Block, PressMethod);
false ->
Block
end,
case fetchend_rawblock(BlockList) of
{LastKey, _LV} when StartKey > LastKey ->
{Acc, true};
{LastKey, _LV} ->
{_LDrop, RKeep} = lists:splitwith(LTrimFun,
BlockList),
case leveled_codec:endkey_passed(EndKey, LastKey) of
true ->
{LKeep, _RDrop} = lists:splitwith(RTrimFun, RKeep),
{Acc ++ LKeep, false};
false ->
{Acc ++ RKeep, true}
end;
_ ->
{Acc, true}
end;
{_ , false} ->
{Acc, false}
end
end,
{Acc, _Continue} = lists:foldl(BlockCheckFun, {[], true}, BlocksToCheck), {Acc, _Continue} = lists:foldl(BlockCheckFun, {[], true}, BlocksToCheck),
Acc. {Acc, none};
{{Header, _Blocks}, SegList} ->
BL = ?BLOCK_LENGTHS_LENGTH,
<<BlockLengths:BL/binary, BlockIdx/binary>> = Header,
PosList = find_pos(BlockIdx, SegList, [], 0),
KVL = check_blocks(PosList,
FullBin,
BlockLengths,
byte_size(BlockIdx),
false,
PressMethod,
[]),
{KVL, Header};
{crc_wonky, _} ->
{[], none}
end.
crc_check_slot(FullBin) -> crc_check_slot(FullBin) ->
@ -2267,7 +2316,9 @@ indexed_list_allindexkeys_test() ->
% "Indexed list flattened in ~w microseconds ~n", % "Indexed list flattened in ~w microseconds ~n",
% [timer:now_diff(os:timestamp(), SW)]), % [timer:now_diff(os:timestamp(), SW)]),
?assertMatch(Keys, BinToList), ?assertMatch(Keys, BinToList),
?assertMatch(Keys, binaryslot_trimmedlist(FullBin, all, all, native)). ?assertMatch({Keys, none}, binaryslot_trimmedlist(FullBin,
all, all,
native, false)).
indexed_list_allindexkeys_nolookup_test() -> indexed_list_allindexkeys_nolookup_test() ->
Keys = lists:sublist(lists:ukeysort(1, generate_indexkeys(1000)), Keys = lists:sublist(lists:ukeysort(1, generate_indexkeys(1000)),
@ -2281,7 +2332,9 @@ indexed_list_allindexkeys_nolookup_test() ->
% "Indexed list flattened in ~w microseconds ~n", % "Indexed list flattened in ~w microseconds ~n",
% [timer:now_diff(os:timestamp(), SW)]), % [timer:now_diff(os:timestamp(), SW)]),
?assertMatch(Keys, BinToList), ?assertMatch(Keys, BinToList),
?assertMatch(Keys, binaryslot_trimmedlist(FullBin, all, all, native)). ?assertMatch({Keys, none}, binaryslot_trimmedlist(FullBin,
all, all,
native, false)).
indexed_list_allindexkeys_trimmed_test() -> indexed_list_allindexkeys_trimmed_test() ->
Keys = lists:sublist(lists:ukeysort(1, generate_indexkeys(150)), Keys = lists:sublist(lists:ukeysort(1, generate_indexkeys(150)),
@ -2290,7 +2343,7 @@ indexed_list_allindexkeys_trimmed_test() ->
generate_binary_slot(lookup, Keys, native, no_timing), generate_binary_slot(lookup, Keys, native, no_timing),
EmptySlotSize = ?LOOK_SLOTSIZE - 1, EmptySlotSize = ?LOOK_SLOTSIZE - 1,
?assertMatch(<<_BL:20/binary, EmptySlotSize:8/integer>>, Header), ?assertMatch(<<_BL:20/binary, EmptySlotSize:8/integer>>, Header),
?assertMatch(Keys, binaryslot_trimmedlist(FullBin, ?assertMatch({Keys, none}, binaryslot_trimmedlist(FullBin,
{i, {i,
"Bucket", "Bucket",
{"t1_int", 0}, {"t1_int", 0},
@ -2299,26 +2352,27 @@ indexed_list_allindexkeys_trimmed_test() ->
"Bucket", "Bucket",
{"t1_int", 99999}, {"t1_int", 99999},
null}, null},
native)), native,
false)),
{SK1, _} = lists:nth(10, Keys), {SK1, _} = lists:nth(10, Keys),
{EK1, _} = lists:nth(100, Keys), {EK1, _} = lists:nth(100, Keys),
R1 = lists:sublist(Keys, 10, 91), R1 = lists:sublist(Keys, 10, 91),
O1 = binaryslot_trimmedlist(FullBin, SK1, EK1, native), {O1, none} = binaryslot_trimmedlist(FullBin, SK1, EK1, native, false),
?assertMatch(91, length(O1)), ?assertMatch(91, length(O1)),
?assertMatch(R1, O1), ?assertMatch(R1, O1),
{SK2, _} = lists:nth(10, Keys), {SK2, _} = lists:nth(10, Keys),
{EK2, _} = lists:nth(20, Keys), {EK2, _} = lists:nth(20, Keys),
R2 = lists:sublist(Keys, 10, 11), R2 = lists:sublist(Keys, 10, 11),
O2 = binaryslot_trimmedlist(FullBin, SK2, EK2, native), {O2, none} = binaryslot_trimmedlist(FullBin, SK2, EK2, native, false),
?assertMatch(11, length(O2)), ?assertMatch(11, length(O2)),
?assertMatch(R2, O2), ?assertMatch(R2, O2),
{SK3, _} = lists:nth(?LOOK_SLOTSIZE - 1, Keys), {SK3, _} = lists:nth(?LOOK_SLOTSIZE - 1, Keys),
{EK3, _} = lists:nth(?LOOK_SLOTSIZE, Keys), {EK3, _} = lists:nth(?LOOK_SLOTSIZE, Keys),
R3 = lists:sublist(Keys, ?LOOK_SLOTSIZE - 1, 2), R3 = lists:sublist(Keys, ?LOOK_SLOTSIZE - 1, 2),
O3 = binaryslot_trimmedlist(FullBin, SK3, EK3, native), {O3, none} = binaryslot_trimmedlist(FullBin, SK3, EK3, native, false),
?assertMatch(2, length(O3)), ?assertMatch(2, length(O3)),
?assertMatch(R3, O3). ?assertMatch(R3, O3).
@ -2379,7 +2433,7 @@ indexed_list_mixedkeys_bitflip_test() ->
{SK1, _} = lists:nth(10, Keys), {SK1, _} = lists:nth(10, Keys),
{EK1, _} = lists:nth(20, Keys), {EK1, _} = lists:nth(20, Keys),
O1 = binaryslot_trimmedlist(SlotBin3, SK1, EK1, native), {O1, none} = binaryslot_trimmedlist(SlotBin3, SK1, EK1, native, false),
?assertMatch([], O1), ?assertMatch([], O1),
SlotBin4 = flip_byte(SlotBin, 0, 20), SlotBin4 = flip_byte(SlotBin, 0, 20),
@ -2391,8 +2445,8 @@ indexed_list_mixedkeys_bitflip_test() ->
ToList5 = binaryslot_tolist(SlotBin5, native), ToList5 = binaryslot_tolist(SlotBin5, native),
?assertMatch([], ToList4), ?assertMatch([], ToList4),
?assertMatch([], ToList5), ?assertMatch([], ToList5),
O4 = binaryslot_trimmedlist(SlotBin4, SK1, EK1, native), {O4, none} = binaryslot_trimmedlist(SlotBin4, SK1, EK1, native, false),
O5 = binaryslot_trimmedlist(SlotBin4, SK1, EK1, native), {O5, none} = binaryslot_trimmedlist(SlotBin4, SK1, EK1, native, false),
?assertMatch([], O4), ?assertMatch([], O4),
?assertMatch([], O5). ?assertMatch([], O5).

59
test/end_to_end/basic_SUITE.erl
View file

@ -270,28 +270,50 @@ fetchput_snapshot(_Config) ->
{ok, Bookie1} = leveled_bookie:book_start(StartOpts1), {ok, Bookie1} = leveled_bookie:book_start(StartOpts1),
{TestObject, TestSpec} = testutil:generate_testobject(), {TestObject, TestSpec} = testutil:generate_testobject(),
ok = testutil:book_riakput(Bookie1, TestObject, TestSpec), ok = testutil:book_riakput(Bookie1, TestObject, TestSpec),
% Load up 5000 objects
ObjList1 = testutil:generate_objects(5000, 2), ObjList1 = testutil:generate_objects(5000, 2),
testutil:riakload(Bookie1, ObjList1), testutil:riakload(Bookie1, ObjList1),
% Now take a snapshot - check it has the same objects
SnapOpts1 = [{snapshot_bookie, Bookie1}], SnapOpts1 = [{snapshot_bookie, Bookie1}],
{ok, SnapBookie1} = leveled_bookie:book_start(SnapOpts1), {ok, SnapBookie1} = leveled_bookie:book_start(SnapOpts1),
ChkList1 = lists:sublist(lists:sort(ObjList1), 100), ChkList1 = lists:sublist(lists:sort(ObjList1), 100),
testutil:check_forlist(Bookie1, ChkList1), testutil:check_forlist(Bookie1, ChkList1),
testutil:check_forlist(SnapBookie1, ChkList1), testutil:check_forlist(SnapBookie1, ChkList1),
% Close the snapshot, check the original store still has the objects
ok = leveled_bookie:book_close(SnapBookie1), ok = leveled_bookie:book_close(SnapBookie1),
testutil:check_forlist(Bookie1, ChkList1), testutil:check_forlist(Bookie1, ChkList1),
ok = leveled_bookie:book_close(Bookie1), ok = leveled_bookie:book_close(Bookie1),
io:format("Closed initial bookies~n"), io:format("Closed initial bookies~n"),
% all now closed
% Open a new store (to start with the previously loaded data)
{ok, Bookie2} = leveled_bookie:book_start(StartOpts1), {ok, Bookie2} = leveled_bookie:book_start(StartOpts1),
SnapOpts2 = [{snapshot_bookie, Bookie2}], SnapOpts2 = [{snapshot_bookie, Bookie2}],
% And take a snapshot of that store
{ok, SnapBookie2} = leveled_bookie:book_start(SnapOpts2), {ok, SnapBookie2} = leveled_bookie:book_start(SnapOpts2),
io:format("Bookies restarted~n"), io:format("Bookies restarted~n"),
% Check both the newly opened store and its snapshot have the data
testutil:check_forlist(Bookie2, ChkList1), testutil:check_forlist(Bookie2, ChkList1),
io:format("Check active bookie still contains original data~n"), io:format("Check active bookie still contains original data~n"),
testutil:check_forlist(SnapBookie2, ChkList1), testutil:check_forlist(SnapBookie2, ChkList1),
io:format("Check snapshot still contains original data~n"), io:format("Check snapshot still contains original data~n"),
% Generate some replacement objects, load them up - check the master
% store has the replacement objects, but the snapshot still has the old
% objects
ObjList2 = testutil:generate_objects(5000, 2), ObjList2 = testutil:generate_objects(5000, 2),
testutil:riakload(Bookie2, ObjList2), testutil:riakload(Bookie2, ObjList2),
@ -303,14 +325,23 @@ fetchput_snapshot(_Config) ->
io:format("Checked for replacement objects in active bookie" ++ io:format("Checked for replacement objects in active bookie" ++
", old objects in snapshot~n"), ", old objects in snapshot~n"),
% Check out how many ledger files we now have (should just be 1)
ok = filelib:ensure_dir(RootPath ++ "/ledger/ledger_files"), ok = filelib:ensure_dir(RootPath ++ "/ledger/ledger_files"),
{ok, FNsA} = file:list_dir(RootPath ++ "/ledger/ledger_files"), {ok, FNsA} = file:list_dir(RootPath ++ "/ledger/ledger_files"),
% generate some new objects and load them up. Check that the master store
% has the new objects, and the snapshot doesn't
ObjList3 = testutil:generate_objects(15000, 5002), ObjList3 = testutil:generate_objects(15000, 5002),
testutil:riakload(Bookie2, ObjList3), testutil:riakload(Bookie2, ObjList3),
ChkList3 = lists:sublist(lists:sort(ObjList3), 100), ChkList3 = lists:sublist(lists:sort(ObjList3), 100),
testutil:check_forlist(Bookie2, ChkList3), testutil:check_forlist(Bookie2, ChkList3),
testutil:check_formissinglist(SnapBookie2, ChkList3), testutil:check_formissinglist(SnapBookie2, ChkList3),
GenList = [20002, 40002, 60002, 80002],
% Now loads lots of new objects
GenList = [20002, 40002, 60002, 80002, 100002],
CLs2 = testutil:load_objects(20000, GenList, Bookie2, TestObject, CLs2 = testutil:load_objects(20000, GenList, Bookie2, TestObject,
fun testutil:generate_smallobjects/2), fun testutil:generate_smallobjects/2),
io:format("Loaded significant numbers of new objects~n"), io:format("Loaded significant numbers of new objects~n"),
@ -318,6 +349,9 @@ fetchput_snapshot(_Config) ->
testutil:check_forlist(Bookie2, lists:nth(length(CLs2), CLs2)), testutil:check_forlist(Bookie2, lists:nth(length(CLs2), CLs2)),
io:format("Checked active bookie has new objects~n"), io:format("Checked active bookie has new objects~n"),
% Start a second snapshot, which should have the new objects, whilst the
% previous snapshot still doesn't
{ok, SnapBookie3} = leveled_bookie:book_start(SnapOpts2), {ok, SnapBookie3} = leveled_bookie:book_start(SnapOpts2),
testutil:check_forlist(SnapBookie3, lists:nth(length(CLs2), CLs2)), testutil:check_forlist(SnapBookie3, lists:nth(length(CLs2), CLs2)),
testutil:check_formissinglist(SnapBookie2, ChkList3), testutil:check_formissinglist(SnapBookie2, ChkList3),
@ -327,6 +361,8 @@ fetchput_snapshot(_Config) ->
testutil:check_forlist(SnapBookie2, ChkList1), testutil:check_forlist(SnapBookie2, ChkList1),
io:format("Started new snapshot and check for new objects~n"), io:format("Started new snapshot and check for new objects~n"),
% Load yet more objects, these are replacement objects for the last load
CLs3 = testutil:load_objects(20000, GenList, Bookie2, TestObject, CLs3 = testutil:load_objects(20000, GenList, Bookie2, TestObject,
fun testutil:generate_smallobjects/2), fun testutil:generate_smallobjects/2),
testutil:check_forlist(Bookie2, lists:nth(length(CLs3), CLs3)), testutil:check_forlist(Bookie2, lists:nth(length(CLs3), CLs3)),
@ -334,22 +370,41 @@ fetchput_snapshot(_Config) ->
io:format("Starting 15s sleep in which snap2 should block deletion~n"), io:format("Starting 15s sleep in which snap2 should block deletion~n"),
timer:sleep(15000), timer:sleep(15000),
% There should be lots of ledger files, as we have replaced the objects
% which has created new files, but the old files are still in demand from
% the snapshot
{ok, FNsB} = file:list_dir(RootPath ++ "/ledger/ledger_files"), {ok, FNsB} = file:list_dir(RootPath ++ "/ledger/ledger_files"),
ok = leveled_bookie:book_close(SnapBookie2), ok = leveled_bookie:book_close(SnapBookie2),
io:format("Starting 15s sleep as snap2 close should unblock deletion~n"), io:format("Starting 15s sleep as snap2 close should unblock deletion~n"),
timer:sleep(15000), timer:sleep(15000),
io:format("Pause for deletion has ended~n"), io:format("Pause for deletion has ended~n"),
% So the pause here is to allow for delete pendings to take effect after the
% closing of the snapshot
% Now check that any deletions haven't impacted the availability of data
testutil:check_forlist(Bookie2, lists:nth(length(CLs3), CLs3)), testutil:check_forlist(Bookie2, lists:nth(length(CLs3), CLs3)),
% Close the other snapshot, and pause - after the pause there should be a
% reduction in the number of ledger files due to the deletes
ok = leveled_bookie:book_close(SnapBookie3), ok = leveled_bookie:book_close(SnapBookie3),
io:format("Starting 15s sleep as snap3 close should unblock deletion~n"), io:format("Starting 15s sleep as snap3 close should unblock deletion~n"),
timer:sleep(15000), timer:sleep(15000),
io:format("Pause for deletion has ended~n"), io:format("Pause for deletion has ended~n"),
testutil:check_forlist(Bookie2, lists:nth(length(CLs3), CLs3)), testutil:check_forlist(Bookie2, lists:nth(length(CLs3), CLs3)),
testutil:check_forlist(Bookie2, lists:nth(1, CLs3)), testutil:check_forlist(Bookie2, lists:nth(1, CLs3)),
{ok, FNsC} = file:list_dir(RootPath ++ "/ledger/ledger_files"), {ok, FNsC} = file:list_dir(RootPath ++ "/ledger/ledger_files"),
io:format("FNsA ~w FNsB ~w FNsC ~w~n",
[length(FNsA), length(FNsB), length(FNsC)]),
true = length(FNsB) > length(FNsA), true = length(FNsB) > length(FNsA),
true = length(FNsB) > length(FNsC), true = length(FNsB) > length(FNsC),
% smaller due to replacements and files deleting
% This is dependent on the sleep though (yuk)
{B1Size, B1Count} = testutil:check_bucket_stats(Bookie2, "Bucket1"), {B1Size, B1Count} = testutil:check_bucket_stats(Bookie2, "Bucket1"),
true = B1Size > 0, true = B1Size > 0,
@ -357,7 +412,7 @@ fetchput_snapshot(_Config) ->
{B1Size, B1Count} = testutil:check_bucket_stats(Bookie2, "Bucket1"), {B1Size, B1Count} = testutil:check_bucket_stats(Bookie2, "Bucket1"),
{BSize, BCount} = testutil:check_bucket_stats(Bookie2, "Bucket"), {BSize, BCount} = testutil:check_bucket_stats(Bookie2, "Bucket"),
true = BSize > 0, true = BSize > 0,
true = BCount == 100000, true = BCount == 120000,
ok = leveled_bookie:book_close(Bookie2), ok = leveled_bookie:book_close(Bookie2),
testutil:reset_filestructure(). testutil:reset_filestructure().

67
test/end_to_end/riak_SUITE.erl
View file

@ -61,6 +61,38 @@ crossbucket_aae(_Config) ->
%% Check all the objects are found - used to trigger HEAD performance log %% Check all the objects are found - used to trigger HEAD performance log
ok = testutil:checkhead_forlist(Bookie2, lists:nth(1, CLs)), ok = testutil:checkhead_forlist(Bookie2, lists:nth(1, CLs)),
test_segfilter_query(Bookie2, CLs),
% Start a new store, and load the same objects (except fot the original
% test object) into this store
%
% This is now the comparison part of the test
StartOpts3 = [{root_path, RootPathB},
{max_journalsize, 200000000},
{max_pencillercachesize, 16000},
{sync_strategy, testutil:sync_strategy()}],
{ok, Bookie3} = leveled_bookie:book_start(StartOpts3),
lists:foreach(fun(ObjL) -> testutil:riakload(Bookie3, ObjL) end, CLs),
test_singledelta_stores(Bookie2, Bookie3, small, {B1, K1}),
test_singledelta_stores(Bookie2, Bookie3, medium, {B1, K1}),
test_singledelta_stores(Bookie2, Bookie3, xsmall, {B1, K1}),
test_singledelta_stores(Bookie2, Bookie3, xxsmall, {B1, K1}),
% Test with a newly opened book (i.e with no block indexes cached)
ok = leveled_bookie:book_close(Bookie2),
{ok, Bookie2A} = leveled_bookie:book_start(StartOpts2),
test_segfilter_query(Bookie2A, CLs),
test_segfilter_query(Bookie2A, CLs),
test_singledelta_stores(Bookie2A, Bookie3, small, {B1, K1}),
ok = leveled_bookie:book_close(Bookie2A),
ok = leveled_bookie:book_close(Bookie3).
test_segfilter_query(Bookie, CLs) ->
% This part of the test tests an issue with accelerating folds by segment % This part of the test tests an issue with accelerating folds by segment
% list, when there is more than one key with a matching segment in the % list, when there is more than one key with a matching segment in the
% slot. Previously this was not handled correctly - and this test part % slot. Previously this was not handled correctly - and this test part
@ -113,16 +145,16 @@ crossbucket_aae(_Config) ->
end, end,
{async, SL1Folder} = {async, SL1Folder} =
leveled_bookie:book_returnfolder(Bookie2, leveled_bookie:book_returnfolder(Bookie,
HeadSegmentFolderGen(SL1, BK1)), HeadSegmentFolderGen(SL1, BK1)),
{async, SL2Folder} = {async, SL2Folder} =
leveled_bookie:book_returnfolder(Bookie2, leveled_bookie:book_returnfolder(Bookie,
HeadSegmentFolderGen(SL2, BK2)), HeadSegmentFolderGen(SL2, BK2)),
{async, SL3Folder} = {async, SL3Folder} =
leveled_bookie:book_returnfolder(Bookie2, leveled_bookie:book_returnfolder(Bookie,
HeadSegmentFolderGen(SL3, BK3)), HeadSegmentFolderGen(SL3, BK3)),
{async, SL4Folder} = {async, SL4Folder} =
leveled_bookie:book_returnfolder(Bookie2, leveled_bookie:book_returnfolder(Bookie,
HeadSegmentFolderGen(SL4, BK4)), HeadSegmentFolderGen(SL4, BK4)),
Results = [SL1Folder(), SL2Folder(), SL3Folder(), SL4Folder()], Results = [SL1Folder(), SL2Folder(), SL3Folder(), SL4Folder()],
@ -130,32 +162,7 @@ crossbucket_aae(_Config) ->
"for SliceSize ~w in ~w ms~n", "for SliceSize ~w in ~w ms~n",
[Results, SliceSize, [Results, SliceSize,
timer:now_diff(os:timestamp(), SW0)/1000]), timer:now_diff(os:timestamp(), SW0)/1000]),
lists:foreach(fun(R) -> true = R == SliceSize end, Results), lists:foreach(fun(R) -> true = R == SliceSize end, Results).
% Start a new store, and load the same objects (except fot the original
% test object) into this store
%
% This is now the comparison part of the test
StartOpts3 = [{root_path, RootPathB},
{max_journalsize, 200000000},
{max_pencillercachesize, 16000},
{sync_strategy, testutil:sync_strategy()}],
{ok, Bookie3} = leveled_bookie:book_start(StartOpts3),
lists:foreach(fun(ObjL) -> testutil:riakload(Bookie3, ObjL) end, CLs),
test_singledelta_stores(Bookie2, Bookie3, small, {B1, K1}),
test_singledelta_stores(Bookie2, Bookie3, medium, {B1, K1}),
test_singledelta_stores(Bookie2, Bookie3, xsmall, {B1, K1}),
test_singledelta_stores(Bookie2, Bookie3, xxsmall, {B1, K1}),
% Test with a newly opend book (i.e with no block indexes cached)
ok = leveled_bookie:book_close(Bookie2),
{ok, Bookie2A} = leveled_bookie:book_start(StartOpts2),
test_singledelta_stores(Bookie2A, Bookie3, small, {B1, K1}),
ok = leveled_bookie:book_close(Bookie2A),
ok = leveled_bookie:book_close(Bookie3).
test_singledelta_stores(BookA, BookB, TreeSize, DeltaKey) -> test_singledelta_stores(BookA, BookB, TreeSize, DeltaKey) ->