rocksdb pr #55 merged into master

Reviewed-on: #9

rebar.config

@@ -4,7 +4,7 @@
 {deps,
  [
   {sext, "1.8.0"},
-  {rocksdb, {git, "https://github.com/emqx/erlang-rocksdb.git", {ref,"d695c6e"}}},
+  {rocksdb, {git, "https://github.com/emqx/erlang-rocksdb.git", {ref, "09df04e"}}},
   {hut, "1.4.0"}
  ]}.
 
@@ -26,7 +26,10 @@
            , {meck, "0.9.2"}
            , {trace_runner, {git, "https://github.com/uwiger/trace_runner.git",
                              {ref, "2e56677"}}}
-           ]}
+           ]},
+    {dialyzer, [{plt_apps, all_deps},
+                {base_plt_apps, [erts, kernel, stdlib, mnesia, runtime_tools, eunit,
+                                proper, trace_runner, common_test]}]}
    ]},
   {edown,
    %% Use as `rebar3 as edown do edoc`

rebar.lock

@@ -2,7 +2,7 @@
 [{<<"hut">>,{pkg,<<"hut">>,<<"1.4.0">>},0},
  {<<"rocksdb">>,
   {git,"https://github.com/emqx/erlang-rocksdb.git",
-       {ref,"d695c6ee9dd27bfe492ed4e24c72ad20ab0d770b"}},
+       {ref,"09df04e02f250a5075bb93cafdd3b637cab81d96"}},
   0},
  {<<"sext">>,{pkg,<<"sext">>,<<"1.8.0">>},0}]}.
 [

src/mnesia_rocksdb_admin.erl

@@ -78,7 +78,7 @@
              | {add_aliases, [alias()]}
              | {write_table_property, tabname(), tuple()}
              | {remove_aliases, [alias()]}
-             | {migrate, [{tabname(), map()}], rpt()}
+             | {migrate, [tabname() | {tabname(), map()}], rpt()}
              | {prep_close, table()}
              | {close_table, table()}
 	     | {clear_table, table() | cf() }.
@@ -272,9 +272,16 @@ delete_info_(Ref, Tab, K) ->
 write_table_property(Alias, Tab, Prop) when is_tuple(Prop), size(Prop) >= 1 ->
     call(Alias, {write_table_property, Tab, Prop}).
 
+-spec migrate_standalone(alias(), Tabs) -> Res when
+      Tabs :: [tabname() | {tabname(),map()}],
+      Res  :: [{tabname(), {ok,any()} | {error, any()}}].
 migrate_standalone(Alias, Tabs) ->
     migrate_standalone(Alias, Tabs, undefined).
 
+-spec migrate_standalone(alias(), Tabs, Rpt) -> Res when
+      Tabs :: [tabname() | {tabname(),map()}],
+      Rpt  :: 'undefined' | pid() | atom(),
+      Res  :: [{tabname(), {ok,any()} | {error, any()}}].
 migrate_standalone(Alias, Tabs, Rpt0) ->
     Rpt = case Rpt0 of
               undefined -> undefined;

src/mrdb.erl

@@ -68,6 +68,7 @@
         , select/2        , select/3
         , select_reverse/2, select_reverse/3
         , select/1
+        , ms/2
         , fold/3            , fold/4        , fold/5
         , fold_reverse/3    , fold_reverse/4, fold_reverse/5
         , rdb_fold/4        , rdb_fold/5
@@ -98,12 +99,18 @@
                             , encode_val/2
                             , decode_val/3 ]).
 
+%% TODO: OTP 28.3 dialyzer is quite pesky about opaque types. Find a way
+%% to structure this, and firm up the type specs.
+-dialyzer([no_opaque, no_return]).
+-dialyzer([{nowarn_function, [activity/3]}]).
+
 -export_type( [ mrdb_iterator/0
               , itr_handle/0
               , iterator_action/0
               , db_ref/0
               , ref_or_tab/0
               , index_position/0
+              , match_pattern/0
               ]).
 
 -include("mnesia_rocksdb.hrl").
@@ -125,6 +132,9 @@
 -type outer() :: non_neg_integer().
 -type retries() :: outer() | {inner(), outer()}.
 
+-type matchpat_map() :: #{atom() => _}.
+-type match_pattern() :: matchpat_map() | ets:match_pattern().
+
 %% activity type 'ets' makes no sense in this context
 -type mnesia_activity_type() :: transaction
                               | sync_transaction
@@ -132,14 +142,15 @@
                               | sync_dirty.
 
 -type tx_options() :: #{ retries => retries()
-                       , no_snapshot => boolean() }.
+                       , no_snapshot => boolean()
+                       , mnesia_compatible => boolean() }.
 -type mrdb_activity_type() :: tx | {tx, tx_options()} | batch.
 
 -type activity_type() :: mrdb_activity_type() | mnesia_activity_type().
 
 -type key() :: any().
 -type obj() :: tuple().
--type index_position() :: atom() | pos().
+-type index_position() :: atom() | pos() | plugin_ix_pos().
 
 -type db_handle()  :: rocksdb:db_handle().
 -type cf_handle()  :: rocksdb:cf_handle().
@@ -155,10 +166,14 @@
 -type activity() :: tx_activity() | batch_activity().
 
 -type pos() :: non_neg_integer().
+-type plugin_ix_pos() :: {atom()}.
 
+-type propkey()    :: any().
+-type propvalue()  :: any().
 -type properties() :: #{ record_name := atom()
                        , attributes  := [atom()]
                        , index       := [{pos(), bag | ordered}]
+                       , user_properties => #{propkey() => propvalue()}
                        }.
 -type semantics()  :: bag | set.
 -type key_encoding() :: 'raw' | 'sext' | 'term'.
@@ -169,7 +184,7 @@
 -type attr_pos() :: #{atom() := pos()}.
 
 -type db_ref() :: #{ name       => table()
-                   , alias      => atom()
+                   , alias      => alias()
                    , vsn        => non_neg_integer()
                    , db_ref     := db_handle()
                    , cf_handle  := cf_handle()
@@ -184,6 +199,11 @@
                    , activity   => activity()
                    , _ => _}.
 
+-type context() :: #{ 'activity' => activity()
+                    , 'alias'    => alias()
+                    , 'retries'  => retries()
+                    , 'db_ref'   => db_ref() }.
+
 -type error() :: {error, any()}.
 
 -type ref_or_tab() :: table() | db_ref().
@@ -275,7 +295,7 @@ release_snapshot(SHandle) ->
 %% To simplify code adaptation, `tx | transaction | sync_transaction' are synonyms, and
 %% `batch | async_dirty | sync_dirty' are synonyms.
 %% @end
--spec activity(activity_type(), alias(), fun( () -> Res )) -> Res.
+-spec activity(activity_type(), alias(), fun( () -> any() )) -> any().
 activity(Type, Alias, F) ->
     #{db_ref := DbRef} = ensure_ref({admin, Alias}),
     Ctxt = case tx_type(Type) of
@@ -293,6 +313,7 @@ activity(Type, Alias, F) ->
            end,
     do_activity(F, Alias, Ctxt).
 
+-spec do_activity(fun(() -> any()), alias(), context()) -> any().
 do_activity(F, Alias, Ctxt) ->
     try try_f(F, Ctxt)
     catch
@@ -309,11 +330,11 @@ try_f(false, F, Ctxt) ->
             commit_and_pop(Res)
     catch
         throw:Something ->
-            abort_and_pop(throw, Something);
+            abort_and_pop(false, throw, Something);
         Cat:Err:T ->
             %% Without capturing the stacktract here,
             %% debugging gets pretty difficult. Incompatible with mnesia, though.
-            abort_and_pop(Cat, {Err, T})
+            abort_and_pop(false, Cat, {Err, T})
     end;
 try_f(true, F, Ctxt) ->
     try run_f(F, Ctxt) of
@@ -321,11 +342,11 @@ try_f(true, F, Ctxt) ->
             commit_and_pop(Res)
     catch
         throw:Something ->
-            abort_and_pop(throw, Something);
+            abort_and_pop(true, throw, Something);
         Cat:Err ->
             %% Without capturing the stacktract here,
             %% debugging gets pretty difficult
-            abort_and_pop(Cat, Err)
+            abort_and_pop(true, Cat, Err)
     end.
 
 
@@ -358,7 +379,7 @@ retry_activity(F, Alias, #{activity := #{ type := Type
                     retry_activity(F, Alias, Ctxt1)
             end;
         error ->
-            return_abort(Type, error, retry_limit)
+            return_abort(mnesia_compatible_aborts(Ctxt), Type, error, retry_limit)
     end.
 
 retry_activity_(inner, F, Alias, Ctxt) ->
@@ -415,6 +436,7 @@ current_context() ->
             undefined
     end.
 
+-spec tx_type(activity_type()) -> {'tx', map()} | 'batch'.
 tx_type(T) ->
     case T of
         _ when T==batch;
@@ -501,8 +523,8 @@ commit_and_pop(Res) ->
             end
     end.
 
--spec abort_and_pop(atom(), any()) -> no_return().
-abort_and_pop(Cat, Err) ->
+-spec abort_and_pop(boolean(), atom(), any()) -> no_return().
+abort_and_pop(Compat, Cat, Err) ->
     %% We can pop the context right away, since there is no
     %% complex failure handling (like retry-on-busy) for rollback.
     #{activity := #{type := Type, handle := H}} = pop_ctxt(),
@@ -510,23 +532,29 @@ abort_and_pop(Cat, Err) ->
         tx    -> ok = rdb_transaction_rollback(H);
         batch -> ok = release_batches(H)
     end,
-    return_abort(Type, Cat, Err).
+    return_abort(Compat, Type, Cat, Err).
 
--spec return_abort(batch | tx, atom(), any()) -> no_return().
-return_abort(batch, Cat, Err) ->
+-spec return_abort(boolean(), batch | tx, atom(), any()) -> no_return().
+return_abort(_, batch, Cat, Err) ->
     re_throw(Cat, Err);
-return_abort(tx, Cat, Err) ->
-    case mnesia_compatible_aborts() of
+return_abort(Compat, tx, Cat, Err) ->
+    case Compat of
 	true ->
 	    %% Mnesia always captures stack traces, but this could actually become a
 	    %% performance issue in some cases (generally, it's better not to lazily
-	    %% produce stack traces.) Since we want to pushe the option checking for
+	    %% produce stack traces.) Since we want to push the option checking for
 	    %% mnesia-abort-style compatibility to AFTER detecting an abort, we don't
 	    %% order a stack trace initially, and instead insert an empty list.
 	    %% (The exact stack trace wouldn't be the same anyway.)
+            %% NOTE: This behavior changed in mnesia-4.23.5.1, so if we really want
+            %% to stay compatible, we have to special-case things.
 	    Err1 =
 		case Cat of
-		    error -> {fix_error(Err), []};
+		    error ->
+                        case newer_mnesia() of
+                            true  -> fix_error(Err);
+                            false -> {fix_error(Err), []}
+                        end;
 		    exit  -> fix_error(Err);
 		    throw -> {throw, Err}
 		end,
@@ -535,6 +563,11 @@ return_abort(tx, Cat, Err) ->
 	    re_throw(Cat, Err)
     end.
 
+%% Some rollback return values changed in mnesia in vsn 4.23.5.1
+newer_mnesia() ->
+    {ok, Vsn} = application:get_key(mnesia, vsn),
+    ("4.23.5" < Vsn).
+
 -spec re_throw(atom(), any()) -> no_return().
 re_throw(Cat, Err) ->
     case Cat of
@@ -1327,6 +1360,55 @@ select_reverse(Tab, Pat, Limit) when Limit == infinity; is_integer(Limit), Limit
 select(Cont) ->
     mrdb_select:select(Cont).
 
+%% @doc Produce a match specification for select(), supporting map-based match patterns
+%%
+%% Using record syntax in match patterns tends to conflict with type checking. This
+%% function offers an alternative approach, drawing on the fact that mnesia_rocksdb
+%% keeps the record name and attribute names readily available as persistent terms.
+%%
+%% When using the map-based representation, the match pattern is built by matching
+%% attribute names to map elements; any attribute not found in the map gets set to '_'.
+%% Thus, ```[{#balance{key = {Acct,'$1'},_='_'},[{'>=','$1',Height}],['$_']}]''' can be
+%% created as ```ms(balance,[{#{key => {Acct,'$1'}},[{'>=','$1',Height}],['$_']}])'''.
+%%
+%% This has the advantage over `ms_transform' that it can handle bound variables
+%% in the match pattern. The function works on all mnesia table types.
+%% @end
+-spec ms(ref_or_tab(), [{match_pattern(), [_], [_]}]) -> ets:match_spec().
+ms(Tab, Pat) ->
+    #{ attributes := Attrs
+     , record_name := RecName } = any_tab_props(Tab),
+    [{headpat(RecName, Attrs, Hd), Gs, Body}
+     || {Hd, Gs, Body} <- Pat].
+
+any_tab_props(Tab) ->
+    try mrdb_props(Tab)
+    catch
+        exit:{aborted,{bad_type,_}} ->
+            mnesia_props(Tab)
+    end.
+
+mrdb_props(Tab) ->
+    #{properties := Props} = get_ref(Tab),
+    Props.
+
+mnesia_props(Tab) ->
+    try mnesia_props_(Tab)
+    catch
+        exit:{aborted, _} ->
+            mnesia:abort({bad_type, Tab})
+    end.
+
+mnesia_props_(Tab) ->
+    #{ record_name => mnesia:table_info(Tab, record_name)
+     , attributes  => mnesia:table_info(Tab, attributes) }.
+
+headpat(RecName, Attrs, Hd) when is_map(Hd) ->
+    list_to_tuple([RecName | [maps:get(A, Hd, '_')
+                              || A <- Attrs]]);
+headpat(_, _, Hd) when is_tuple(Hd); is_atom(Hd) ->
+    Hd.
+
 clear_table(Tab) ->
     match_delete(Tab, '_').
 

src/mrdb_index.erl

@@ -9,6 +9,10 @@
         , fold/4
         , rev_fold/4
         , index_ref/2
+        , select/3
+        , select/4
+        , select_reverse/3
+        , select_reverse/4
         ]).
 
 -record(mrdb_ix_iter, { i    :: mrdb:mrdb_iterator()
@@ -33,6 +37,11 @@
 
 -export_type([ ix_iterator/0 ]).
 
+-spec index_ref(mrdb:ref_or_tab(), mrdb:index_position()) -> mrdb:db_ref().
+index_ref(Tab, Ix) ->
+    #{} = R = mrdb:ensure_ref(Tab),
+    ensure_index_ref(Ix, R).
+
 -spec with_iterator(mrdb:ref_or_tab(), mrdb:index_position(), fun( (ix_iterator()) -> Res)) -> Res.
 with_iterator(Tab, IxPos, Fun) when is_function(Fun, 1) ->
     {ok, I} = iterator(Tab, IxPos),
@@ -88,6 +97,26 @@ fold_(Tab, IxPos, Start, Dir, FoldFun, Acc) ->
               iter_fold(I, Start, Dir, FoldFun, Acc)
       end).
 
+select(Tab, Ix, MS) ->
+    select(Tab, Ix, MS, infinity).
+
+select(Tab, Ix, MS, Limit) ->
+    #{} = R = mrdb:ensure_ref(Tab),
+    #{} = IxR = ensure_index_ref(Ix, R),
+    MSpre = pre_match_spec(MS),
+    mrdb_select:select(IxR#{ derive_obj_f => mk_derive_obj_f(R)
+                           , pre_ms => MSpre }, MS, Limit).
+
+select_reverse(Tab, Ix, MS) ->
+    select_reverse(Tab, Ix, MS, infinity).
+
+select_reverse(Tab, Ix, MS, Limit) ->
+    #{} = R = mrdb:ensure_ref(Tab),
+    #{} = IxR = ensure_index_ref(Ix, R),
+    MSpre = pre_match_spec(MS),
+    mrdb_select:select_reverse(IxR#{ derive_obj_f => mk_derive_obj_f(R)
+                                   , pre_ms => MSpre }, MS, Limit).
+
 iter_fold(I, Start, Dir, Fun, Acc) ->
     iter_fold_(iterator_move(I, Start), I, Dir, Fun, Acc).
 
@@ -96,10 +125,6 @@ iter_fold_({ok, IxVal, Obj}, I, Dir, Fun, Acc) ->
 iter_fold_({error, _}, _, _, _, Acc) ->
     Acc.
 
-index_ref(Tab, Pos) ->
-    TRef = mrdb:ensure_ref(Tab),
-    ensure_index_ref(Pos, TRef).
-
 iterator_move_set(#mrdb_ix_iter{i = I, sub = Sub}, Dir) ->
     case mrdb:iterator_move(I, Dir) of
         {ok, {{FKey, PKey}}} ->
@@ -121,6 +146,29 @@ iterator_move_bag(#mrdb_ix_iter{i = I, sub = Sub}, Dir) ->
             Other
     end.
 
+pre_match_spec([{{KeyPat,ObjPat}, Gs, MatchBody} | T]) ->
+    [{{{KeyPat,'_'},ObjPat}, Gs, MatchBody} | pre_match_spec(T)];
+pre_match_spec([H|T]) ->
+    [H | pre_match_spec(T)];
+pre_match_spec([]) ->
+    [].
+
+%% Used for mrdb_select:select()
+%% The select operation folds over index keys, and for matching keys,
+%% calls the `derive_obj_f/1` fun, which normally just does `Obj -> [Obj]`.
+%% In the case of indexes, it fetches the object, if it exists, and then
+%% returns `[{IndexKey, Object}]`, which is what the matchspec should operate on.
+%%
+mk_derive_obj_f(Ref) ->
+    fun({{IxK, Key}}) ->
+            case mrdb:read(Ref, Key, []) of
+                [Obj] ->
+                    [{IxK, Obj}];
+                [] ->
+                    []
+            end
+    end.
+
 -spec opt_read(mrdb:ref_or_tab(), Key :: any()) -> any().
 opt_read(R, Key) ->
     case mrdb:read(R, Key, []) of

src/mrdb_select.erl

@@ -30,6 +30,8 @@
              , limit
              , key_only = false                 % TODO: not used
              , direction = forward
+             , pre_ms
+             , derive_obj_f = fun unit_l/1
              }).
 
 select(Ref, MS, Limit) when is_map(Ref), is_list(MS) ->
@@ -50,15 +52,18 @@ select(Ref, MS, AccKeys, Dir, Limit)
     mrdb:with_rdb_iterator(Ref, fun(I) -> i_select(I, Sel, AccKeys, []) end).
 
 mk_sel(#{name := Tab} = Ref, MS, Dir, Limit) ->
-    Keypat = keypat(MS, keypos(Tab), Ref),
+    MSpre = maps:get(pre_ms, Ref, MS),
+    Keypat = keypat(MSpre, keypos(Tab), Ref),
     #sel{tab = Tab,
          ref = Ref,
          keypat = Keypat,
-         ms = MS,
-         compiled_ms = ets:match_spec_compile(MS),
+         ms = MSpre,
+         pre_ms = MSpre,
+         compiled_ms = ms_compile(MS),
          key_only = needs_key_only(MS),
          direction = Dir,
-         limit = Limit}.
+         limit = Limit,
+         derive_obj_f = derive_f(Ref)}.
 
 select(Cont) ->
     case Cont of
@@ -73,6 +78,7 @@ continuation_info(_, _) -> undefined.
 
 continuation_info_(ref,       #sel{ref       = Ref}) -> Ref;
 continuation_info_(ms,        #sel{ms        = MS }) -> MS;
+continuation_info_(pre_ms,    #sel{pre_ms    = MSp}) -> MSp;
 continuation_info_(limit,     #sel{limit     = L  }) -> L;
 continuation_info_(direction, #sel{direction = Dir}) -> Dir;
 continuation_info_(_, _) -> undefined.
@@ -124,7 +130,21 @@ fwd_init_seek_tgt(<<>>  ) -> first;
 fwd_init_seek_tgt(Prefix) -> {seek, Prefix}.
 
 rev_init_seek(I, Pfx) ->
-    rocksdb:iterator_move(I, rev_init_seek_tgt(Pfx)).
+    case rev_init_seek_tgt(Pfx) of
+        last ->
+            i_move(I, last);
+        {seek, Bin} ->
+            %% An 'incremented' prefix.
+            %% This will fail if we seek past the end of the table.
+            %% Then, try to seek_for_prev instead (fails if table empty).
+            %% This because rocksdb lacks a "seek backward to last matching prefix".
+            case i_move(I, {seek, Bin}) of
+                {error, invalid_iterator} ->
+                    i_move(I, {seek_for_prev, Bin});
+                {ok, _, _} = Ok ->
+                    Ok
+            end
+    end.
 
 rev_init_seek_tgt(<<>>) -> last;
 rev_init_seek_tgt(Prefix) ->
@@ -173,19 +193,19 @@ i_select(I, #sel{ keypat = Pfx
                 , limit = Limit
                 , direction = Dir
                 , ref = #{encoding := Enc} } = Sel0, AccKeys, Acc) ->
-    {StartKey, Sel} = case Enc of
-                          {term, _} ->
+    {MoveRes, Sel} = case Enc of
+                         {term, _} ->
                               %% No defined ordering - do forward select
-                              {first, Sel0#sel{direction = forward}};
-                          _ ->
-                              SK = case Dir of
-                                       forward -> fwd_init_seek_tgt(Pfx);
-                                       reverse -> rev_init_seek_tgt(Pfx)
-                                   end,
-                              {SK, Sel0}
-                      end,
-    select_traverse(rocksdb:iterator_move(I, StartKey), Limit,
-                    Pfx, MS, I, Sel, AccKeys, Acc).
+                             {i_move(I, first), Sel0#sel{direction = forward}};
+                         _ ->
+                             case Dir of
+                                 forward ->
+                                     {fwd_init_seek(I, Pfx), Sel0};
+                                 reverse ->
+                                     {rev_init_seek(I, Pfx), Sel0}
+                             end
+                     end,
+    select_traverse(MoveRes, Limit, Pfx, MS, I, Sel, AccKeys, Acc).
 
 needs_key_only([Pat]) ->
     needs_key_only_(Pat);
@@ -266,8 +286,9 @@ select_traverse({ok, K, V}, Limit, Pfx, MS, I, #sel{ref = R, direction = Dir} =
     case is_prefix(Pfx, K) of
         true ->
             DecKey = decode_key(K, R),
-            Rec = decode_val(V, DecKey, R),
-            case ets:match_spec_run([Rec], MS) of
+            Rec0 = decode_val(V, DecKey, R),
+            RecL = derive_object(Rec0, Sel),
+            case ms_run(RecL, MS) of
                 [] ->
                     select_traverse(
                       rocksdb:iterator_move(I, next_or_prev(Dir)), Limit, Pfx, MS,
@@ -336,11 +357,17 @@ iterator_next(I, K, Dir) ->
             Other
     end.
 
+i_move(I, Tgt) ->
+    rocksdb:iterator_move(I, Tgt).
+
 i_move(I, K, reverse) ->
     rocksdb:iterator_move(I, {seek_for_prev, K});
 i_move(I, K, forward) ->
     rocksdb:iterator_move(I, K).
 
+derive_object(R, #sel{derive_obj_f = F}) ->
+    F(R).
+
 keypat([H|T], KeyPos, Ref) ->
     keypat(T, KeyPos, Ref, keypat_pfx(H, KeyPos, Ref)).
 
@@ -362,3 +389,14 @@ keypat_pfx({HeadPat,_Gs,_}, KeyPos, #{encoding := {sext,_}}) when is_tuple(HeadP
 keypat_pfx(_, _, _) ->
     <<>>.
 
+derive_f(#{derive_obj_f := F}) when is_function(F, 1) -> F;
+derive_f(_) -> fun unit_l/1.
+
+unit_l(X) ->
+    [X].
+
+ms_compile(MS) ->
+  ets:match_spec_compile(MS).
+
+ms_run(RecL, MS) ->
+    ets:match_spec_run(RecL, MS).

test/basho_bench_driver_mnesia_rocksdb.erl

@@ -1,64 +0,0 @@
-%% -------------------------------------------------------------------
-%%
-%% basho_bench: Benchmarking Suite
-%%
-%% Copyright (c) 2009-2010 Basho Techonologies
-%%
-%% This file is provided to you under the Apache License,
-%% Version 2.0 (the "License"); you may not use this file
-%% except in compliance with the License.  You may obtain
-%% a copy of the License at
-%%
-%%   http://www.apache.org/licenses/LICENSE-2.0
-%%
-%% Unless required by applicable law or agreed to in writing,
-%% software distributed under the License is distributed on an
-%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-%% KIND, either express or implied.  See the License for the
-%% specific language governing permissions and limitations
-%% under the License.
-%%
-%% -------------------------------------------------------------------
-
--module(basho_bench_driver_mnesia_rocksdb).
-
--export([new/1,
-         run/4]).
-
--include("mnesia_rocksdb_basho_bench.hrl").
-
-%% ====================================================================
-%% API
-%% ====================================================================
-
-new(_Id) ->
-    Type = basho_bench_config:get(backend, ram_copies),
-    Tab = basho_bench_config:get(mnesia_table, t),
-    ok = bootstrap_mnesia(Tab, Type),
-    {ok, Tab}.
-
-bootstrap_mnesia(Tab, Type) ->
-    ok = mnesia:create_schema([node()],
-                              [{backend_types,
-                                [{rocksdb_copies, mnesia_rocksdb}]}]),
-    ok = mnesia:start(),
-    {atomic,ok} = mnesia:create_table(Tab, [{Type, [node()]}]),
-    mnesia:wait_for_tables([Tab], 10000).
-
-run(get, KeyGen, _ValueGen, State) ->
-    Tab = State,
-    Key = KeyGen(),
-    case mnesia:dirty_read({Tab, Key}) of
-        [] ->
-            {ok, State};
-        [{_, Key, _}] ->
-            {ok, State}
-    end;
-run(put, KeyGen, ValueGen, State) ->
-    Tab = State,
-    ok = mnesia:dirty_write({Tab, KeyGen(), ValueGen()}),
-    {ok, State};
-run(delete, KeyGen, _ValueGen, State) ->
-    Tab = State,
-    ok = mnesia:dirty_delete({Tab, KeyGen()}),
-    {ok, State}.

test/mnesia_rocksdb_SUITE.erl

@@ -216,6 +216,7 @@ mrdb_transactions_(Config) ->
     delete_tabs(Created),
     ok.
 
+-dialyzer(no_return).
 mrdb_abort_reasons(_Config) ->
     Prev = mnesia_rocksdb_admin:set_and_cache_env(mnesia_compatible_aborts, true),
     X = some_value,
@@ -234,6 +235,7 @@ compare_txs(Type, F) ->
     ct:log("Mrdb   = ~p/~p", [Type, EMr]),
     case {Type, EMn, EMr} of
 	{error, {some_value, [_|_]}, {some_value, []}} -> ok;
+        {error, E, E} -> ok;
 	{throw, {throw, some_value}, {throw, some_value}} -> ok;
 	{exit, some_value, some_value} -> ok;
 	{abort, some_value, some_value} -> ok
@@ -276,21 +278,30 @@ mrdb_abort(Config) ->
     mrdb:insert(tx_abort, {tx_abort, a, 1}),
     Pre = mrdb:read(tx_abort, a),
     D0 = get_dict(),
+    ActivityF = fun() ->
+                        [{tx_abort, a, N}] = mrdb:read(tx_abort, a),
+                        error(abort_here),
+                        ok = mrdb:insert(tx_abort, [{tx_abort, a, N+1}]),
+                        noooo
+                end,
     TRes = try mrdb:activity(
                  {tx, #{mnesia_compatible => true}}, rdb,
-                 fun() ->
-                         [{tx_abort, a, N}] = mrdb:read(tx_abort, a),
-                         error(abort_here),
-                         ok = mrdb:insert(tx_abort, [{tx_abort, a, N+1}]),
-                         noooo
-                 end)
+                ActivityF)
            catch
-               error:abort_here ->
+               exit:{aborted, {abort_here, []}} ->
                    ok
            end,
     dictionary_unchanged(D0),
     ok = TRes,
     Pre = mrdb:read(tx_abort, a),
+    TRes1 = try mrdb:activity(tx, rdb, ActivityF)
+                     catch
+                         error:{abort_here, [_|_]} ->
+                             ok
+                     end,
+    dictionary_unchanged(D0),
+    ok = TRes1,
+    Pre = mrdb:read(tx_abort, a),
     delete_tabs(Created),
     ok.
 

test/mnesia_rocksdb_basho_bench.hrl

@@ -1,15 +0,0 @@
-
--define(FAIL_MSG(Str, Args), ?ERROR(Str, Args), basho_bench_app:halt_or_kill()).
--define(STD_ERR(Str, Args), io:format(standard_error, Str, Args)).
-
--define(CONSOLE(Str, Args), lager:info(Str, Args)).
-
--define(DEBUG(Str, Args), lager:debug(Str, Args)).
--define(INFO(Str, Args), lager:info(Str, Args)).
--define(WARN(Str, Args), lager:warning(Str, Args)).
--define(ERROR(Str, Args), lager:error(Str, Args)).
-
--define(FMT(Str, Args), lists:flatten(io_lib:format(Str, Args))).
-
--define(VAL_GEN_BLOB_CFG, value_generator_blob_file).
--define(VAL_GEN_SRC_SIZE, value_generator_source_size).

test/mnesia_rocksdb_bench_disc_only.config

@@ -1,18 +0,0 @@
-{mode, max}.
-
-{duration, 10}.
-
-{concurrent, 1}.
-
-{driver, basho_bench_driver_mnesia_rocksdb}.
-
-{key_generator, {int_to_bin,{uniform_int, 5000000}}}.
-
-{value_generator, {fixed_bin, 10000}}.
-
-{operations, [{get, 2}, {put, 2}, {delete, 1}]}.
-
-{code_paths, []}.
-
-{mnesia_table, doc}.
-{backend, disc_only_copies}.

test/mnesia_rocksdb_bench_rocksdb_copies.config

@@ -1,19 +0,0 @@
-{mode, max}.
-
-{duration, 10}.
-
-{concurrent, 1}.
-
-{driver, basho_bench_driver_mnesia_rocksdb}.
-
-{key_generator, {int_to_bin,{uniform_int, 5000000}}}.
-
-{value_generator, {fixed_bin, 10000}}.
-
-{operations, [{get, 2}, {put, 2}, {delete, 1}]}.
-
-{code_paths, ["/Users/uwiger/git/rocksdb",
-              "/Users/uwiger/git/mnesia_rocksdb"]}.
-
-{mnesia_table, rdb}.
-{backend, rocksdb_copies}.

test/mnesia_rocksdb_conv_bigtab.erl

@@ -22,6 +22,7 @@
 
 -record(t, {k, i, v}).
 
+-dialyzer({nowarn_function, run/1}).
 run(Sz) ->
     mnesia:stop(),
     init(),

test/mnesia_rocksdb_fallback.erl

@@ -79,20 +79,20 @@ cleanup() ->
     os:cmd("rm *.BUP").
 
 mods(0) ->
-    [];
-mods(1) ->
-    [
-     {l, mnesia_rocksdb},
-     {g, rocksdb}
-    ];
-mods(2) ->
-    [
-     %% {l, mnesia_monitor},
-     {g, mnesia_rocksdb},
-     {l, mnesia_bup},
-     {g, mnesia_lib},
-     {g, mnesia_schema},
-     %% {g, mnesia_loader},
-     {g, mnesia_index},
-     {l, mnesia_tm}
-    ].
+    [].
+%% mods(1) ->
+%%     [
+%%      {l, mnesia_rocksdb},
+%%      {g, rocksdb}
+%%     ];
+%% mods(2) ->
+%%     [
+%%      %% {l, mnesia_monitor},
+%%      {g, mnesia_rocksdb},
+%%      {l, mnesia_bup},
+%%      {g, mnesia_lib},
+%%      {g, mnesia_schema},
+%%      %% {g, mnesia_loader},
+%%      {g, mnesia_index},
+%%      {l, mnesia_tm}
+%%     ].

test/mnesia_rocksdb_indexes_SUITE.erl

@@ -152,8 +152,8 @@ end_per_testcase(_, _) ->
 %% ======================================================================
 
 cfg([Tab, Type, IxType], Config) ->
-    [{my_config, #{tab => Tab, type => Type, ixtype => IxType}} | Config];
-cfg(Cfg, Config) when is_map(Cfg) -> [{my_config, Cfg} | Config].
+    [{my_config, #{tab => Tab, type => Type, ixtype => IxType}} | Config].
+%% cfg(Cfg, Config) when is_map(Cfg) -> [{my_config, Cfg} | Config].
 
 cfg(Config) -> ?config(my_config, Config).
 
@@ -236,12 +236,25 @@ test_index_plugin(Config) ->
     if Type == rdb ->
             Res1 = lists:sort(mrdb:index_read(Tab,<<"sen">>, {pfx})),
             Res2 = lists:sort(mrdb:index_read(Tab,<<"whi">>, {pfx})),
+            ok = test_select(Tab,{pfx},[{'_', [], ['$_']}]),
+            MS2 = [{{<<"whi">>,{Tab,"truth",'_'}},[],['$_']}],
+            [_] = mrdb_index:select(Tab, {pfx}, MS2),
+            ok = test_select(Tab, {pfx}, MS2),
             [{Tab,"foobar","sentence"}] = mrdb:index_read(
                                             Tab, <<"foo">>, {pfx});
        true ->
             ok
     end.
 
+test_select(Tab, Ix, MS) ->
+    Res = mrdb_index:select(Tab, Ix, MS),
+    ct:log("mrdb_index:select(~p, ~p, ~p) -> ~p", [Tab, Ix, MS, Res]),
+    RevRes = mrdb_index:select_reverse(Tab, Ix, MS),
+    ct:log("mrdb_index:select_reverse(~p, ~p, ~p) -> ~p", [Tab, Ix, MS, RevRes]),
+    {Res,Res} = {Res, lists:reverse(RevRes)},
+    ok.
+
+
 ixtype(T) when T==bag;
                T==ordered ->
     {{pfx}, T};

test/mnesia_rocksdb_proper_semantics_test.erl

@@ -48,14 +48,18 @@
 -record(st, {}).
 -define(KEYS, [a,b,c]).
 
+-dialyzer({no_return, [basic_test_/0, test/1, prop_seq/0]}).
+
 basic_test_() ->
     {timeout, 60000, [fun() -> test(100) end]}.
 
+-dialyzer({no_opaque, test/1}).
 test(N) ->
     setup_mnesia(),
     true = proper:quickcheck(?MODULE:prop_seq(), N),
     ok.
 
+-dialyzer({no_opaque, prop_seq/0}).
 prop_seq() ->
     ?FORALL(Cmds, proper_statem:commands(?MODULE),
             begin

test/mnesia_rocksdb_xform.erl

@@ -260,10 +260,10 @@ plain_transform1(Fun, [F|Fs]) when is_atom(element(1,F)) ->
         continue ->
             [list_to_tuple(plain_transform1(Fun, tuple_to_list(F))) |
              plain_transform1(Fun, Fs)];
-        {done, NewF} ->
-            [NewF | Fs];
-        {error, Reason} ->
-            io:format("Error: ~p (~p)~n", [F,Reason]);
+        %% {done, NewF} ->
+        %%     [NewF | Fs];
+        %% {error, Reason} ->
+        %%     io:format("Error: ~p (~p)~n", [F,Reason]);
         NewF when is_tuple(NewF) ->
             [NewF | plain_transform1(Fun, Fs)]
     end;

test/mrdb_fold_SUITE.erl

@@ -90,7 +90,7 @@ mk_tab(G) ->
     end.
 
 tab_name(fold    ) -> r;
-tab_name(select  ) -> r;
+%% tab_name(select  ) -> r;
 tab_name(rdb_fold) -> t.
 
 create_tab(T) ->