Merge pull request 'Ensure the erlang_merge_operator is always loaded' (#5) from uw-merge-operator into master

Reviewed-on: #5

src/mnesia_rocksdb_admin.erl

@@ -1489,24 +1489,26 @@ open_db_(MP, Alias, Opts, CFs0, CreateIfMissing) ->
             %% not yet created
             CFs = cfs(CFs0),
             file:make_dir(MP),
-            OpenOpts = [ {create_if_missing, true}
+            OpenRes = rocksdb_open(MP, Opts, CFs),
-                       , {create_missing_column_families, true}
-                       , {merge_operator, erlang_merge_operator}
-                       | Opts ],
-            OpenRes = mnesia_rocksdb_lib:open_rocksdb(MP, OpenOpts, CFs),
             map_cfs(OpenRes, CFs, Alias, Acc0);
         false ->
             {error, enoent};
         true ->
             %% Assumption: even an old rocksdb database file will have at least "default"
             {ok,CFs} = rocksdb:list_column_families(MP, Opts),
-            CFs1 = [{CF,[]} || CF <- CFs], %% TODO: this really needs more checking
+            CFs1 = [{CF, cfopts()} || CF <- CFs], %% TODO: this really needs more checking
             map_cfs(rocksdb_open(MP, Opts, CFs1), CFs1, Alias, Acc0)
     end.
 
+open_opts(Opts) ->
+    [ {create_if_missing, true}
+    , {create_missing_column_families, true}
+    , {merge_operator, erlang_merge_operator}
+    | Opts ].
+
 rocksdb_open(MP, Opts, CFs) ->
     %% rocksdb:open(MP, Opts, CFs),
-    mnesia_rocksdb_lib:open_rocksdb(MP, Opts, CFs).
+    mnesia_rocksdb_lib:open_rocksdb(MP, open_opts(Opts), CFs).
 
 is_open(Alias, #st{backends = Bs}) ->
     case maps:find(Alias, Bs) of

src/mrdb.erl

@@ -53,6 +53,7 @@
         , delete/2       , delete/3
         , delete_object/2, delete_object/3
         , match_delete/2
+        , merge/3        , merge/4
 	, clear_table/1
         , batch_write/2  , batch_write/3
         , update_counter/3, update_counter/4
@@ -740,6 +741,21 @@ insert(Tab, Obj0, Opts) ->
     EncVal = encode_val(Obj, Ref),
     insert_(Ref, Key, encode_key(Key, Ref), EncVal, Obj, Opts).
 
+merge(Tab, Key, MergeOp) ->
+    merge(Tab, Key, MergeOp, []).
+
+merge(Tab, Key, MergeOp, Opts) ->
+    #{encoding := Enc} = Ref = ensure_ref(Tab),
+    case Enc of
+        {_, {value, term}} ->
+            merge_(Ref, Key, MergeOp, Opts);
+        _ ->
+            abort(badarg)
+    end.
+
+merge_(Ref, Key, MergeOp, Opts) ->
+    rdb_merge(Ref, encode_key(Key), MergeOp, Opts).
+
 validate_obj(Obj, #{mode := mnesia}) ->
     Obj;
 validate_obj(Obj, #{attr_pos := AP,

test/mnesia_rocksdb_SUITE.erl

@@ -27,6 +27,9 @@
         , mrdb_two_procs_tx_inner_restart/1
         , mrdb_two_procs_snap/1
         , mrdb_three_procs/1
+        , create_counters/1
+        , update_counters/1
+        , restart_node/1
         ]).
 
 -include_lib("common_test/include/ct.hrl").
@@ -42,7 +45,8 @@ all() ->
 groups() ->
     [
       {all_tests, [sequence], [ {group, checks}
-                              , {group, mrdb} ]}
+                              , {group, mrdb}
+                              , {group, counters}]}
                               %% , error_handling ]}
     , {checks, [sequence], [ encoding_sext_attrs
                            , encoding_binary_binary
@@ -57,6 +61,10 @@ groups() ->
                          , mrdb_two_procs_tx_inner_restart
                          , mrdb_two_procs_snap
                          , mrdb_three_procs ]}
+    , {counters, [sequence], [ create_counters
+                             , update_counters
+                             , restart_node
+                             , update_counters ]}
     ].
 
 
@@ -749,6 +757,35 @@ get_attempt() ->
     #{activity := #{attempt := Attempt}} = mrdb:current_context(),
     Attempt.
 
+create_counters(_Config) ->
+    create_tab(counters, []),
+    mrdb:insert(counters, {counters, c0, 1}),
+    mrdb:update_counter(counters, c1, 1),
+    [{counters, c0, 1}] = mrdb:read(counters, c0),
+    [{counters, c1, 1}] = mrdb:read(counters, c1),
+    ct:log("Created tab counters, with objs c0 (1) and c1 (1)", []),
+    ok.
+
+restart_node(_Config) ->
+    mnesia:stop(),
+    ok = mnesia:start(),
+    ct:log("mnesia restarted", []),
+    ok.
+
+update_counters(_Config) ->
+    [{counters, c0, C0Prev}] = mrdb:read(counters, c0),
+    [{counters, c1, C1Prev}] = mrdb:read(counters, c1),
+    ct:log("co: ~p, c1: ~p", [C0Prev, C1Prev]),
+    ok = mrdb:update_counter(counters, c0, 1),
+    ok = mrdb:update_counter(counters, c1, 1),
+    ct:log("Incremented c0 and c1 by 1", []),
+    C0 = C0Prev + 1,
+    C1 = C1Prev + 1,
+    [{counters, c0, C0}] = mrdb:read(counters, c0),
+    [{counters, c1, C1}] = mrdb:read(counters, c1),
+    ct:log("c0: ~p, c1: ~p", [C0, C1]),
+    ok.
+
 create_tabs(Tabs, Config) ->
     Res = lists:map(fun create_tab/1, Tabs),
     tr_ct:trace_checkpoint(?TABS_CREATED, Config),