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

Support for 2i query part1

Browse source 
Added basic support for 2i query.  This involved some refactoring of the
test code to share functions between suites.

There is sill a need for a Part 2 as no tests currently cover removal of
index entries.
This commit is contained in:
martinsumner 2016-10-18 01:59:03 +01:00
parent ac0504e79e
commit 3e475f46e8
11 changed files with 682 additions and 288 deletions
Download patch file Download diff file Expand all files Collapse all files

89
src/leveled_bookie.erl
View file

@ -124,7 +124,7 @@
-behaviour(gen_server).
-include("../include/leveled.hrl").
-include("include/leveled.hrl").
-export([init/1,
handle_call/3,
@ -348,6 +348,17 @@ handle_call({return_folder, FolderType}, _From, State) ->
State#state.ledger_cache,
Bucket,
?RIAK_TAG),
State};
{index_query,
Bucket,
{IdxField, StartValue, EndValue},
{ReturnTerms, TermRegex}} ->
{reply,
index_query(State#state.penciller,
State#state.ledger_cache,
Bucket,
{IdxField, StartValue, EndValue},
{ReturnTerms, TermRegex}),
State}
end;
handle_call({compact_journal, Timeout}, _From, State) ->
@ -408,6 +419,41 @@ bucket_stats(Penciller, LedgerCache, Bucket, Tag) ->
end,
{async, Folder}.
index_query(Penciller, LedgerCache,
Bucket,
{IdxField, StartValue, EndValue},
{ReturnTerms, TermRegex}) ->
PCLopts = #penciller_options{start_snapshot=true,
source_penciller=Penciller},
{ok, LedgerSnapshot} = leveled_penciller:pcl_start(PCLopts),
Folder = fun() ->
Increment = gb_trees:to_list(LedgerCache),
io:format("Length of increment in snapshot is ~w~n",
[length(Increment)]),
ok = leveled_penciller:pcl_loadsnapshot(LedgerSnapshot,
{infinity, Increment}),
StartKey = leveled_codec:to_ledgerkey(Bucket, null, ?IDX_TAG,
IdxField, StartValue),
EndKey = leveled_codec:to_ledgerkey(Bucket, null, ?IDX_TAG,
IdxField, EndValue),
AddFun = case ReturnTerms of
true ->
fun add_terms/3;
_ ->
fun add_keys/3
end,
AccFun = accumulate_index(TermRegex, AddFun),
Acc = leveled_penciller:pcl_fetchkeys(LedgerSnapshot,
StartKey,
EndKey,
AccFun,
[]),
ok = leveled_penciller:pcl_close(LedgerSnapshot),
Acc
end,
{async, Folder}.
shutdown_wait([], _Inker) ->
false;
shutdown_wait([TopPause|Rest], Inker) ->
@ -476,6 +522,47 @@ accumulate_size(Key, Value, {Size, Count}) ->
{Size, Count}
end.
add_keys(ObjKey, _IdxValue, Acc) ->
Acc ++ [ObjKey].
add_terms(ObjKey, IdxValue, Acc) ->
Acc ++ [{IdxValue, ObjKey}].
accumulate_index(TermRe, AddFun) ->
case TermRe of
undefined ->
fun(Key, Value, Acc) ->
case leveled_codec:is_active(Key, Value) of
true ->
{_Bucket,
ObjKey,
IdxValue} = leveled_codec:from_ledgerkey(Key),
AddFun(ObjKey, IdxValue, Acc);
false ->
Acc
end end;
TermRe ->
fun(Key, Value, Acc) ->
case leveled_codec:is_active(Key, Value) of
true ->
{_Bucket,
ObjKey,
IdxValue} = leveled_codec:from_ledgerkey(Key),
case re:run(IdxValue, TermRe) of
nomatch ->
Acc;
_ ->
AddFun(ObjKey, IdxValue, Acc)
end;
false ->
Acc
end end
end.
preparefor_ledgercache(PK, SQN, Obj, Size, IndexSpecs) ->
{Bucket, Key, PrimaryChange} = leveled_codec:generate_ledgerkv(PK,
SQN,

2
src/leveled_cdb.erl
View file

@ -44,7 +44,7 @@
-module(leveled_cdb).
-behaviour(gen_server).
-include("../include/leveled.hrl").
-include("include/leveled.hrl").
-export([init/1,
handle_call/3,

26
src/leveled_codec.erl
View file

@ -28,7 +28,7 @@
-module(leveled_codec).
-include("../include/leveled.hrl").
-include("include/leveled.hrl").
-include_lib("eunit/include/eunit.hrl").
@ -43,14 +43,24 @@
key_dominates/2,
print_key/1,
to_ledgerkey/3,
to_ledgerkey/5,
from_ledgerkey/1,
build_metadata_object/2,
generate_ledgerkv/4,
generate_ledgerkv/5,
get_size/2,
convert_indexspecs/4,
riakto_keydetails/1]).
riakto_keydetails/1,
generate_uuid/0]).
%% Credit to
%% https://github.com/afiskon/erlang-uuid-v4/blob/master/src/uuid.erl
generate_uuid() ->
<<A:32, B:16, C:16, D:16, E:48>> = crypto:rand_bytes(16),
io_lib:format("~8.16.0b-~4.16.0b-4~3.16.0b-~4.16.0b-~12.16.0b",
[A, B, C band 16#0fff, D band 16#3fff bor 16#8000, E]).
strip_to_keyonly({keyonly, K}) -> K;
strip_to_keyonly({K, _V}) -> K.
@ -87,6 +97,13 @@ is_active(Key, Value) ->
false
end.
from_ledgerkey({Tag, Bucket, {_IdxField, IdxValue}, Key})
when Tag == ?IDX_TAG ->
{Bucket, Key, IdxValue}.
to_ledgerkey(Bucket, Key, Tag, Field, Value) when Tag == ?IDX_TAG ->
{?IDX_TAG, Bucket, {Field, Value}, Key}.
to_ledgerkey(Bucket, Key, Tag) ->
{Tag, Bucket, Key, null}.
@ -132,7 +149,7 @@ endkey_passed(EndKey, CheckingKey) ->
EndKey < CheckingKey.
convert_indexspecs(IndexSpecs, Bucket, Key, SQN) ->
lists:map(fun({IndexOp, IndexField, IndexValue}) ->
lists:map(fun({IndexOp, IdxField, IdxValue}) ->
Status = case IndexOp of
add ->
%% TODO: timestamp support
@ -141,7 +158,8 @@ convert_indexspecs(IndexSpecs, Bucket, Key, SQN) ->
%% TODO: timestamps for delayed reaping
{tomb, infinity}
end,
{{i, Bucket, {IndexField, IndexValue}, Key},
{to_ledgerkey(Bucket, Key, ?IDX_TAG,
IdxField, IdxValue),
{SQN, Status, null}}
end,
IndexSpecs).

2
src/leveled_iclerk.erl
View file

@ -4,7 +4,7 @@
-behaviour(gen_server).
-include("../include/leveled.hrl").
-include("include/leveled.hrl").
-export([init/1,
handle_call/3,

15
src/leveled_inker.erl
View file

@ -76,7 +76,7 @@
-behaviour(gen_server).
-include("../include/leveled.hrl").
-include("include/leveled.hrl").
-export([init/1,
handle_call/3,
@ -669,20 +669,14 @@ filepath(RootPath, journal_compact_dir) ->
filepath(RootPath, NewSQN, new_journal) ->
filename:join(filepath(RootPath, journal_dir),
integer_to_list(NewSQN) ++ "_"
++ generate_uuid()
++ leveled_codec:generate_uuid()
++ "." ++ ?PENDING_FILEX);
filepath(CompactFilePath, NewSQN, compact_journal) ->
filename:join(CompactFilePath,
integer_to_list(NewSQN) ++ "_"
++ generate_uuid()
++ leveled_codec:generate_uuid()
++ "." ++ ?PENDING_FILEX).
%% Credit to
%% https://github.com/afiskon/erlang-uuid-v4/blob/master/src/uuid.erl
generate_uuid() ->
<<A:32, B:16, C:16, D:16, E:48>> = crypto:rand_bytes(16),
io_lib:format("~8.16.0b-~4.16.0b-4~3.16.0b-~4.16.0b-~12.16.0b",
[A, B, C band 16#0fff, D band 16#3fff bor 16#8000, E]).
simple_manifest_reader(SQN, RootPath) ->
ManifestPath = filepath(RootPath, manifest_dir),
@ -815,6 +809,9 @@ simple_inker_completeactivejournal_test() ->
F2 = filename:join(JournalFP, "nursery_3.pnd"),
{ok, PidW} = leveled_cdb:cdb_open_writer(F2),
{ok, _F2} = leveled_cdb:cdb_complete(PidW),
F1 = filename:join(JournalFP, "nursery_1.cdb"),
F1r = filename:join(JournalFP, "nursery_1.pnd"),
ok = file:rename(F1, F1r),
{ok, Ink1} = ink_start(#inker_options{root_path=RootPath,
cdb_options=CDBopts}),
Obj1 = ink_get(Ink1, "Key1", 1),

2
src/leveled_pclerk.erl
View file

@ -52,7 +52,7 @@
-behaviour(gen_server).
-include("../include/leveled.hrl").
-include("include/leveled.hrl").
-export([init/1,
handle_call/3,

2
src/leveled_penciller.erl
View file

@ -223,7 +223,7 @@
-behaviour(gen_server).
-include("../include/leveled.hrl").
-include("include/leveled.hrl").
-export([init/1,
handle_call/3,

2
src/leveled_sft.erl
View file

@ -143,7 +143,7 @@
-module(leveled_sft).
-behaviour(gen_server).
-include("../include/leveled.hrl").
-include("include/leveled.hrl").
-export([init/1,
handle_call/3,