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

Read-only opening

Browse source 
Completing the file also closes it and leads to a read-only opening of
the file.
This commit is contained in:
martinsumner 2016-07-07 11:43:26 +01:00
parent 3b954aea43
commit 9dae893958
1 changed files with 77 additions and 74 deletions
Download patch file Download diff file Expand all files Collapse all files

151
src/leveled_sft.erl
View file

@ -178,20 +178,19 @@
index_pointer :: integer(), index_pointer :: integer(),
filter_pointer :: integer(), filter_pointer :: integer(),
summ_pointer :: integer(), summ_pointer :: integer(),
summ_length :: integer()}). summ_length :: integer(),
filename :: string()}).
%% Start a bare file with an initial header and no further details %% Start a bare file with an initial header and no further details
%% Return the {Handle, metadata record} %% Return the {Handle, metadata record}
create_file(FileName) when is_list(FileName) -> create_file(FileName) when is_list(FileName) ->
{ok, Handle} = file:open(FileName, [binary, raw, read, write]), {ok, Handle} = file:open(FileName, [binary, raw, read, write]),
create_file(Handle);
create_file(Handle) ->
Header = create_header(initial), Header = create_header(initial),
{ok, _} = file:position(Handle, bof), {ok, _} = file:position(Handle, bof),
ok = file:write(Handle, Header), ok = file:write(Handle, Header),
{ok, StartPos} = file:position(Handle, cur), {ok, StartPos} = file:position(Handle, cur),
FileMD = #state{next_position=StartPos}, FileMD = #state{next_position=StartPos, filename=FileName},
{Handle, FileMD}. {Handle, FileMD}.
@ -210,40 +209,42 @@ create_header(initial) ->
%% Open a file returning a handle and metadata which can be used in fetch and %% Open a file returning a handle and metadata which can be used in fetch and
%% iterator requests %% iterator requests
%% The handle should be read-only as these are immutable files, a file cannot
%% be opened for writing keys, it can only be created to write keys
open_file(Filename) -> open_file(FileMD) ->
{ok, _Handle} = file:open(Filename, [binary, raw, read, write]). Filename = FileMD#state.filename,
%% Need to write other metadata somewhere {ok, Handle} = file:open(Filename, [binary, raw, read]),
%% ... probably in summmary {ok, HeaderLengths} = file:pread(Handle, 12, 16),
%% ... is there a need for two levels of summary? <<Blen:32/integer,
Ilen:32/integer,
Flen:32/integer,
Slen:32/integer>> = HeaderLengths,
{ok, SummaryBin} = file:pread(Handle, ?HEADER_LEN + Blen + Ilen + Flen, Slen),
{{LowSQN, HighSQN}, {LowKey, HighKey}} = binary_to_term(SummaryBin),
{ok, SlotIndexBin} = file:pread(Handle, ?HEADER_LEN + Blen, Ilen),
SlotIndex = binary_to_term(SlotIndexBin),
{Handle, FileMD#state{slot_index=SlotIndex,
smallest_sqn=LowSQN,
highest_sqn=HighSQN,
smallest_key=LowKey,
highest_key=HighKey,
slots_pointer=?HEADER_LEN,
index_pointer=?HEADER_LEN + Blen,
filter_pointer=?HEADER_LEN + Blen + Ilen,
summ_pointer=?HEADER_LEN + Blen + Ilen + Flen,
summ_length=Slen}}.
%% Take a file handle with a previously created header and complete it based on %% Take a file handle with a previously created header and complete it based on
%% the two key lists KL1 and KL2 %% the two key lists KL1 and KL2
complete_file(Handle, FileMD, KL1, KL2, Level) -> complete_file(Handle, FileMD, KL1, KL2, Level) ->
{{UpdHandle, {ok, KeyRemainders} = write_keys(Handle,
PointerList, KL1, KL2, [], <<>>,
{LowSQN, HighSQN}, Level,
{LowKey, HighKey}}, fun sftwrite_function/2),
KeyRemainders} = write_group(Handle, KL1, KL2, [], <<>>, Level, {ReadHandle, UpdFileMD} = open_file(FileMD),
fun sftwrite_function/2), {ReadHandle, UpdFileMD, KeyRemainders}.
{ok, HeaderLengths} = file:pread(UpdHandle, 12, 16),
<<Blen:32/integer,
Ilen:32/integer,
Flen:32/integer,
Slen:32/integer>> = HeaderLengths,
{UpdHandle,
FileMD#state{slot_index=PointerList,
smallest_sqn=LowSQN,
highest_sqn=HighSQN,
smallest_key=LowKey,
highest_key=HighKey,
slots_pointer=?HEADER_LEN,
index_pointer=?HEADER_LEN + Blen,
filter_pointer=?HEADER_LEN + Blen + Ilen,
summ_pointer=?HEADER_LEN + Blen + Ilen + Flen,
summ_length=Slen},
KeyRemainders}.
%% Fetch a Key and Value from a file, returns %% Fetch a Key and Value from a file, returns
%% {value, KV} or not_present %% {value, KV} or not_present
@ -259,14 +260,17 @@ fetch_keyvalue(Handle, FileMD, Key) ->
SegID = hash_for_segmentid({keyonly, Key}), SegID = hash_for_segmentid({keyonly, Key}),
case check_for_segments(SegFilter, [SegID], true) of case check_for_segments(SegFilter, [SegID], true) of
{maybe_present, BlockList} -> {maybe_present, BlockList} ->
io:format("Filter has returned maybe~n"),
fetch_keyvalue_fromblock(BlockList, fetch_keyvalue_fromblock(BlockList,
Key, Key,
LengthList, LengthList,
Handle, Handle,
PointerB + FileMD#state.slots_pointer); PointerB + FileMD#state.slots_pointer);
not_present -> not_present ->
io:format("Filter has returned no~n"),
not_present; not_present;
error_so_maybe_present -> error_so_maybe_present ->
io:format("Filter has returned error~n"),
fetch_keyvalue_fromblock(lists:seq(0,3), fetch_keyvalue_fromblock(lists:seq(0,3),
Key, Key,
LengthList, LengthList,
@ -309,30 +313,30 @@ get_nearestkey([Result|T], KeyToFind, _) ->
%% Slots are created then written in bulk to impove I/O efficiency. Slots will %% Slots are created then written in bulk to impove I/O efficiency. Slots will
%% be written in groups of 32 %% be written in groups of 32
write_group(Handle, KL1, KL2, SlotIndex, SerialisedSlots, Level, WriteFun) -> write_keys(Handle, KL1, KL2, SlotIndex, SerialisedSlots, Level, WriteFun) ->
write_group(Handle, KL1, KL2, {0, 0}, write_keys(Handle, KL1, KL2, {0, 0},
SlotIndex, SerialisedSlots, SlotIndex, SerialisedSlots,
{infinity, 0}, null, {last, null}, Level, WriteFun). {infinity, 0}, null, {last, null}, Level, WriteFun).
write_group(Handle, KL1, KL2, {SlotCount, SlotTotal}, write_keys(Handle, KL1, KL2, {SlotCount, SlotTotal},
SlotIndex, SerialisedSlots, SlotIndex, SerialisedSlots,
{LSN, HSN}, LowKey, LastKey, Level, WriteFun) {LSN, HSN}, LowKey, LastKey, Level, WriteFun)
when SlotCount =:= ?SLOT_GROUPWRITE_COUNT -> when SlotCount =:= ?SLOT_GROUPWRITE_COUNT ->
UpdHandle = WriteFun(slots , {Handle, SerialisedSlots}), UpdHandle = WriteFun(slots , {Handle, SerialisedSlots}),
case maxslots_bylevel(SlotTotal, Level) of case maxslots_bylevel(SlotTotal, Level) of
reached -> reached ->
{complete_write(UpdHandle, {complete_keywrite(UpdHandle,
SlotIndex, SlotIndex,
{LSN, HSN}, {LowKey, LastKey}, {LSN, HSN}, {LowKey, LastKey},
WriteFun), WriteFun),
{KL1, KL2}}; {KL1, KL2}};
continue -> continue ->
write_group(UpdHandle, KL1, KL2, {0, SlotTotal}, write_keys(UpdHandle, KL1, KL2, {0, SlotTotal},
SlotIndex, <<>>, SlotIndex, <<>>,
{LSN, HSN}, LowKey, LastKey, Level, WriteFun) {LSN, HSN}, LowKey, LastKey, Level, WriteFun)
end; end;
write_group(Handle, KL1, KL2, {SlotCount, SlotTotal}, write_keys(Handle, KL1, KL2, {SlotCount, SlotTotal},
SlotIndex, SerialisedSlots, SlotIndex, SerialisedSlots,
{LSN, HSN}, LowKey, LastKey, Level, WriteFun) -> {LSN, HSN}, LowKey, LastKey, Level, WriteFun) ->
SlotOutput = create_slot(KL1, KL2, Level), SlotOutput = create_slot(KL1, KL2, Level),
@ -348,28 +352,26 @@ write_group(Handle, KL1, KL2, {SlotCount, SlotTotal},
case Status of case Status of
partial -> partial ->
UpdHandle = WriteFun(slots , {Handle, UpdSlots}), UpdHandle = WriteFun(slots , {Handle, UpdSlots}),
{complete_write(UpdHandle, {complete_keywrite(UpdHandle,
UpdSlotIndex, UpdSlotIndex,
SNExtremes, {FirstKey, FinalKey}, SNExtremes, {FirstKey, FinalKey},
WriteFun), WriteFun),
{KL1rem, KL2rem}}; {KL1rem, KL2rem}};
full -> full ->
write_group(Handle, KL1rem, KL2rem, {SlotCount + 1, SlotTotal + 1}, write_keys(Handle, KL1rem, KL2rem, {SlotCount + 1, SlotTotal + 1},
UpdSlotIndex, UpdSlots, UpdSlotIndex, UpdSlots,
SNExtremes, FirstKey, FinalKey, Level, WriteFun) SNExtremes, FirstKey, FinalKey, Level, WriteFun)
end. end.
complete_write(Handle, SlotIndex, complete_keywrite(Handle, SlotIndex,
SNExtremes, {FirstKey, FinalKey}, SNExtremes, {FirstKey, FinalKey},
WriteFun) -> WriteFun) ->
ConvSlotIndex = convert_slotindex(SlotIndex), ConvSlotIndex = convert_slotindex(SlotIndex),
FinHandle = WriteFun(finalise, {Handle, WriteFun(finalise, {Handle,
ConvSlotIndex, ConvSlotIndex,
SNExtremes, SNExtremes,
{FirstKey, FinalKey}}), {FirstKey, FinalKey}}).
{_, PointerIndex} = ConvSlotIndex,
{FinHandle, PointerIndex, SNExtremes, {FirstKey, FinalKey}}.
%% Take a slot index, and remove the SegFilters replacing with pointers %% Take a slot index, and remove the SegFilters replacing with pointers
@ -398,22 +400,26 @@ sftwrite_function(slots, {Handle, SerialisedSlots}) ->
sftwrite_function(finalise, sftwrite_function(finalise,
{Handle, {Handle,
{SlotFilters, PointerIndex}, {SlotFilters, PointerIndex},
_SNExtremes, SNExtremes,
_KeyExtremes}) -> KeyExtremes}) ->
{ok, Position} = file:position(Handle, cur), {ok, Position} = file:position(Handle, cur),
SlotsLength = Position - ?HEADER_LEN, BlocksLength = Position - ?HEADER_LEN,
SerialisedIndex = term_to_binary(PointerIndex), Index = term_to_binary(PointerIndex),
IndexLength = byte_size(SerialisedIndex), IndexLength = byte_size(Index),
FilterLength = byte_size(SlotFilters),
%% Write Index Summary = term_to_binary({SNExtremes, KeyExtremes}),
ok = file:write(Handle, SerialisedIndex), SummaryLength = byte_size(Summary),
%% Write Filter %% Write Index, Filter and Summary
ok = file:write(Handle, SlotFilters), ok = file:write(Handle, <<Index/binary,
SlotFilters/binary,
Summary/binary>>),
%% Write Lengths into header %% Write Lengths into header
ok = file:pwrite(Handle, 12, <<SlotsLength:32/integer, ok = file:pwrite(Handle, 12, <<BlocksLength:32/integer,
IndexLength:32/integer>>), IndexLength:32/integer,
Handle; FilterLength:32/integer,
SummaryLength:32/integer>>),
file:close(Handle);
sftwrite_function(finalise, sftwrite_function(finalise,
{Handle, {Handle,
SlotIndex, SlotIndex,
@ -1152,12 +1158,10 @@ testwrite_function(slots, {Handle, SerialisedSlots}) ->
testwrite_function(finalise, {Handle, C_SlotIndex, SNExtremes, KeyExtremes}) -> testwrite_function(finalise, {Handle, C_SlotIndex, SNExtremes, KeyExtremes}) ->
{Handle, C_SlotIndex, SNExtremes, KeyExtremes}. {Handle, C_SlotIndex, SNExtremes, KeyExtremes}.
writegroup_stage1_test() -> writekeys_stage1_test() ->
{KL1, KL2} = sample_keylist(), {KL1, KL2} = sample_keylist(),
Output = write_group([], KL1, KL2, [], <<>>, 1, fun testwrite_function/2), {FunOut, {_KL1Rem, _KL2Rem}} = write_keys([], KL1, KL2, [], <<>>, 1, fun testwrite_function/2),
{{{Handle, {_, PointerIndex}, SNExtremes, KeyExtremes}, {Handle, {_, PointerIndex}, SNExtremes, KeyExtremes} = FunOut,
PointerIndex, SNExtremes, KeyExtremes},
{_KL1Rem, _KL2Rem}} = Output,
?assertMatch(SNExtremes, {1,3}), ?assertMatch(SNExtremes, {1,3}),
?assertMatch(KeyExtremes, {{o, "Bucket1", "Key1"}, ?assertMatch(KeyExtremes, {{o, "Bucket1", "Key1"},
{o, "Bucket4", "Key1"}}), {o, "Bucket4", "Key1"}}),
@ -1201,7 +1205,6 @@ big_create_file_test() ->
[{K2, Sq2, St2, V2}|_] = KL2, [{K2, Sq2, St2, V2}|_] = KL2,
Result1 = fetch_keyvalue(Handle, FileMD, K1), Result1 = fetch_keyvalue(Handle, FileMD, K1),
Result2 = fetch_keyvalue(Handle, FileMD, K2), Result2 = fetch_keyvalue(Handle, FileMD, K2),
io:format("Results are:~n~w~n~w~n", [Result1, Result2]),
?assertMatch(Result1, {K1, Sq1, St1, V1}), ?assertMatch(Result1, {K1, Sq1, St1, V1}),
?assertMatch(Result2, {K2, Sq2, St2, V2}), ?assertMatch(Result2, {K2, Sq2, St2, V2}),
FailedFinds = lists:foldl(fun(K, Acc) -> FailedFinds = lists:foldl(fun(K, Acc) ->