From 66e992035d0f74cd8713d4e8ba5044d64613d65b Mon Sep 17 00:00:00 2001 From: Ulf Wiger Date: Thu, 19 Nov 2015 11:42:14 +0100 Subject: mnesia_ext: Add ext copies and db_fold to low level api --- lib/mnesia/src/mnesia_lib.erl | 139 ++++++++++++++++++++++++++++++++++++------ 1 file changed, 121 insertions(+), 18 deletions(-) (limited to 'lib/mnesia') diff --git a/lib/mnesia/src/mnesia_lib.erl b/lib/mnesia/src/mnesia_lib.erl index ca3fc95b0c..10e232c800 100644 --- a/lib/mnesia/src/mnesia_lib.erl +++ b/lib/mnesia/src/mnesia_lib.erl @@ -54,6 +54,7 @@ db_erase_tab/2, db_first/1, db_first/2, + db_foldl/3, db_foldl/4, db_foldl/6, db_last/1, db_last/2, db_fixtable/3, @@ -135,6 +136,7 @@ set_local_content_whereabouts/1, set_remote_where_to_read/1, set_remote_where_to_read/2, + semantics/2, show/1, show/2, sort_commit/1, @@ -144,6 +146,7 @@ tab2tmp/1, tab2dcd/1, tab2dcl/1, + tab2logtmp/1, to_list/1, union/2, uniq/1, @@ -321,15 +324,41 @@ tab2dcd(Tab) -> %% Disc copies data tab2dcl(Tab) -> %% Disc copies log dir(lists:concat([Tab, ".DCL"])). +tab2logtmp(Tab) -> %% Disc copies log + dir(lists:concat([Tab, ".LOGTMP"])). + storage_type_at_node(Node, Tab) -> search_key(Node, [{disc_copies, val({Tab, disc_copies})}, {ram_copies, val({Tab, ram_copies})}, - {disc_only_copies, val({Tab, disc_only_copies})}]). + {disc_only_copies, val({Tab, disc_only_copies})}| + wrap_external(val({Tab, external_copies}))]). cs_to_storage_type(Node, Cs) -> search_key(Node, [{disc_copies, Cs#cstruct.disc_copies}, {ram_copies, Cs#cstruct.ram_copies}, - {disc_only_copies, Cs#cstruct.disc_only_copies}]). + {disc_only_copies, Cs#cstruct.disc_only_copies} | + wrap_external(Cs#cstruct.external_copies)]). + +-define(native(T), T==ram_copies; T==disc_copies; T==disc_only_copies). + +semantics({ext,Alias,Mod}, Item) -> + Mod:semantics(Alias, Item); +semantics({Alias,Mod}, Item) -> + Mod:semantics(Alias, Item); +semantics(Type, storage) when ?native(Type) -> + Type; +semantics(Type, types) when ?native(Type) -> + [set, ordered_set, bag]; +semantics(disc_only_copies, index_types) -> + [bag]; +semantics(Type, index_types) when ?native(Type) -> + [bag, ordered]; +semantics(_, _) -> + undefined. + + +wrap_external(L) -> + [{{ext,Alias,Mod},Ns} || {{Alias,Mod},Ns} <- L]. schema_cs_to_storage_type(Node, Cs) -> case cs_to_storage_type(Node, Cs) of @@ -337,7 +366,6 @@ schema_cs_to_storage_type(Node, Cs) -> Other -> Other end. - search_key(Key, [{Val, List} | Tail]) -> case lists:member(Key, List) of true -> Val; @@ -439,9 +467,9 @@ pr_other(Var) -> no -> {node_not_running, node()}; _ -> {no_exists, Var} end, - verbose("~p (~p) val(mnesia_gvar, ~w) -> ~p ~n", + verbose("~p (~p) val(mnesia_gvar, ~w) -> ~p ~p ~n", [self(), process_info(self(), registered_name), - Var, Why]), + Var, Why, erlang:get_stacktrace()]), mnesia:abort(Why). %% Some functions for list valued variables @@ -578,10 +606,16 @@ read_counter(Name) -> ?ets_lookup_element(mnesia_stats, Name, 2). cs_to_nodes(Cs) -> + ext_nodes(Cs#cstruct.external_copies) ++ Cs#cstruct.disc_only_copies ++ Cs#cstruct.disc_copies ++ Cs#cstruct.ram_copies. +ext_nodes(Ext) -> + lists:flatmap(fun({_, Ns}) -> + Ns + end, Ext). + overload_types() -> [mnesia_tm, mnesia_dump_log]. @@ -777,7 +811,7 @@ view(File) -> true -> view(File, dat); false -> - case suffix([".LOG", ".BUP", ".ETS"], File) of + case suffix([".LOG", ".BUP", ".ETS", ".LOGTMP"], File) of true -> view(File, log); false -> @@ -1073,18 +1107,24 @@ db_get(Tab, Key) -> db_get(val({Tab, storage_type}), Tab, Key). db_get(ram_copies, Tab, Key) -> ?ets_lookup(Tab, Key); db_get(disc_copies, Tab, Key) -> ?ets_lookup(Tab, Key); -db_get(disc_only_copies, Tab, Key) -> dets:lookup(Tab, Key). +db_get(disc_only_copies, Tab, Key) -> dets:lookup(Tab, Key); +db_get({ext, Alias, Mod}, Tab, Key) -> + Mod:lookup(Alias, Tab, Key). db_init_chunk(Tab) -> db_init_chunk(val({Tab, storage_type}), Tab, 1000). db_init_chunk(Tab, N) -> db_init_chunk(val({Tab, storage_type}), Tab, N). +db_init_chunk({ext, Alias, Mod}, Tab, N) -> + Mod:select(Alias, Tab, [{'_', [], ['$_']}], N); db_init_chunk(disc_only_copies, Tab, N) -> dets:select(Tab, [{'_', [], ['$_']}], N); db_init_chunk(_, Tab, N) -> ets:select(Tab, [{'_', [], ['$_']}], N). +db_chunk({ext, _Alias, Mod}, State) -> + Mod:select(State); db_chunk(disc_only_copies, State) -> dets:select(State); db_chunk(_, State) -> @@ -1095,7 +1135,9 @@ db_put(Tab, Val) -> db_put(ram_copies, Tab, Val) -> ?ets_insert(Tab, Val), ok; db_put(disc_copies, Tab, Val) -> ?ets_insert(Tab, Val), ok; -db_put(disc_only_copies, Tab, Val) -> dets:insert(Tab, Val). +db_put(disc_only_copies, Tab, Val) -> dets:insert(Tab, Val); +db_put({ext, Alias, Mod}, Tab, Val) -> + Mod:insert(Alias, Tab, Val). db_match_object(Tab, Pat) -> db_match_object(val({Tab, storage_type}), Tab, Pat). @@ -1104,12 +1146,36 @@ db_match_object(Storage, Tab, Pat) -> try case Storage of disc_only_copies -> dets:match_object(Tab, Pat); + {ext, Alias, Mod} -> Mod:select(Alias, Tab, [{Pat, [], ['$_']}]); _ -> ets:match_object(Tab, Pat) end after db_fixtable(Storage, Tab, false) end. +db_foldl(Fun, Acc, Tab) -> + db_foldl(val({Tab, storage_type}), Fun, Acc, Tab). + +db_foldl(Storage, Fun, Acc, Tab) -> + Limit = mnesia_monitor:get_env(fold_chunk_size), + db_foldl(Storage, Fun, Acc, Tab, [{'_', [], ['$_']}], Limit). + +db_foldl(ram_copies, Fun, Acc, Tab, Pat, Limit) -> + mnesia_lib:db_fixtable(ram_copies, Tab, true), + try select_foldl(db_select_init(ram_copies, Tab, Pat, Limit), + Fun, Acc, ram_copies) + after + mnesia_lib:db_fixtable(ram_copies, Tab, false) + end; +db_foldl(Storage, Fun, Acc, Tab, Pat, Limit) -> + select_foldl(mnesia_lib:db_select_init(Storage, Tab, Pat, Limit), Fun, Acc, Storage). + +select_foldl({Objs, Cont}, Fun, Acc, Storage) -> + select_foldl(mnesia_lib:db_select_cont(Storage, Cont, []), + Fun, lists:foldl(Fun, Acc, Objs), Storage); +select_foldl('$end_of_table', _, Acc, _) -> + Acc. + db_select(Tab, Pat) -> db_select(val({Tab, storage_type}), Tab, Pat). @@ -1118,17 +1184,33 @@ db_select(Storage, Tab, Pat) -> try case Storage of disc_only_copies -> dets:select(Tab, Pat); + {ext, Alias, Mod} -> Mod:select(Alias, Tab, Pat); _ -> ets:select(Tab, Pat) end after db_fixtable(Storage, Tab, false) end. +db_select_init({ext, Alias, Mod}, Tab, Pat, Limit) -> + case Mod:select(Alias, Tab, Pat, Limit) of + {Matches, Continuation} when is_list(Matches) -> + {Matches, {Alias, Continuation}}; + R -> + R + end; db_select_init(disc_only_copies, Tab, Pat, Limit) -> dets:select(Tab, Pat, Limit); db_select_init(_, Tab, Pat, Limit) -> ets:select(Tab, Pat, Limit). +db_select_cont({ext, Alias, Mod}, Cont0, Ms) -> + Cont = Mod:repair_continuation(Cont0, Ms), + case Mod:select(Cont) of + {Matches, Continuation} when is_list(Matches) -> + {Matches, {Alias, Continuation}}; + R -> + R + end; db_select_cont(disc_only_copies, Cont0, Ms) -> Cont = dets:repair_continuation(Cont0, Ms), dets:select(Cont); @@ -1145,13 +1227,18 @@ db_fixtable(disc_copies, Tab, Bool) -> db_fixtable(dets, Tab, Bool) -> dets:safe_fixtable(Tab, Bool); db_fixtable(disc_only_copies, Tab, Bool) -> - dets:safe_fixtable(Tab, Bool). + dets:safe_fixtable(Tab, Bool); +db_fixtable({ext, Alias, Mod}, Tab, Bool) -> + Mod:fixtable(Alias, Tab, Bool). db_erase(Tab, Key) -> db_erase(val({Tab, storage_type}), Tab, Key). db_erase(ram_copies, Tab, Key) -> ?ets_delete(Tab, Key), ok; db_erase(disc_copies, Tab, Key) -> ?ets_delete(Tab, Key), ok; -db_erase(disc_only_copies, Tab, Key) -> dets:delete(Tab, Key). +db_erase(disc_only_copies, Tab, Key) -> dets:delete(Tab, Key); +db_erase({ext, Alias, Mod}, Tab, Key) -> + Mod:delete(Alias, Tab, Key), + ok. db_match_erase(Tab, '_') -> db_delete_all(val({Tab, storage_type}),Tab); @@ -1159,7 +1246,10 @@ db_match_erase(Tab, Pat) -> db_match_erase(val({Tab, storage_type}), Tab, Pat). db_match_erase(ram_copies, Tab, Pat) -> ?ets_match_delete(Tab, Pat), ok; db_match_erase(disc_copies, Tab, Pat) -> ?ets_match_delete(Tab, Pat), ok; -db_match_erase(disc_only_copies, Tab, Pat) -> dets:match_delete(Tab, Pat). +db_match_erase(disc_only_copies, Tab, Pat) -> dets:match_delete(Tab, Pat); +db_match_erase({ext, Alias, Mod}, Tab, Pat) -> + Mod:match_delete(Alias, Tab, Pat), + ok. db_delete_all(ram_copies, Tab) -> ets:delete_all_objects(Tab); db_delete_all(disc_copies, Tab) -> ets:delete_all_objects(Tab); @@ -1169,31 +1259,41 @@ db_first(Tab) -> db_first(val({Tab, storage_type}), Tab). db_first(ram_copies, Tab) -> ?ets_first(Tab); db_first(disc_copies, Tab) -> ?ets_first(Tab); -db_first(disc_only_copies, Tab) -> dets:first(Tab). +db_first(disc_only_copies, Tab) -> dets:first(Tab); +db_first({ext, Alias, Mod}, Tab) -> + Mod:first(Alias, Tab). db_next_key(Tab, Key) -> db_next_key(val({Tab, storage_type}), Tab, Key). db_next_key(ram_copies, Tab, Key) -> ?ets_next(Tab, Key); db_next_key(disc_copies, Tab, Key) -> ?ets_next(Tab, Key); -db_next_key(disc_only_copies, Tab, Key) -> dets:next(Tab, Key). +db_next_key(disc_only_copies, Tab, Key) -> dets:next(Tab, Key); +db_next_key({ext, Alias, Mod}, Tab, Key) -> + Mod:next(Alias, Tab, Key). db_last(Tab) -> db_last(val({Tab, storage_type}), Tab). db_last(ram_copies, Tab) -> ?ets_last(Tab); db_last(disc_copies, Tab) -> ?ets_last(Tab); -db_last(disc_only_copies, Tab) -> dets:first(Tab). %% Dets don't have order +db_last(disc_only_copies, Tab) -> dets:first(Tab); %% Dets don't have order +db_last({ext, Alias, Mod}, Tab) -> + Mod:last(Alias, Tab). db_prev_key(Tab, Key) -> db_prev_key(val({Tab, storage_type}), Tab, Key). db_prev_key(ram_copies, Tab, Key) -> ?ets_prev(Tab, Key); db_prev_key(disc_copies, Tab, Key) -> ?ets_prev(Tab, Key); -db_prev_key(disc_only_copies, Tab, Key) -> dets:next(Tab, Key). %% Dets don't have order +db_prev_key(disc_only_copies, Tab, Key) -> dets:next(Tab, Key); %% Dets don't have order +db_prev_key({ext, Alias, Mod}, Tab, Key) -> + Mod:prev(Alias, Tab, Key). db_slot(Tab, Pos) -> db_slot(val({Tab, storage_type}), Tab, Pos). db_slot(ram_copies, Tab, Pos) -> ?ets_slot(Tab, Pos); db_slot(disc_copies, Tab, Pos) -> ?ets_slot(Tab, Pos); -db_slot(disc_only_copies, Tab, Pos) -> dets:slot(Tab, Pos). +db_slot(disc_only_copies, Tab, Pos) -> dets:slot(Tab, Pos); +db_slot({ext, Alias, Mod}, Tab, Pos) -> + Mod:slot(Alias, Tab, Pos). db_update_counter(Tab, C, Val) -> db_update_counter(val({Tab, storage_type}), Tab, C, Val). @@ -1202,13 +1302,16 @@ db_update_counter(ram_copies, Tab, C, Val) -> db_update_counter(disc_copies, Tab, C, Val) -> ?ets_update_counter(Tab, C, Val); db_update_counter(disc_only_copies, Tab, C, Val) -> - dets:update_counter(Tab, C, Val). + dets:update_counter(Tab, C, Val); +db_update_counter({ext, Alias, Mod}, Tab, C, Val) -> + Mod:update_counter(Alias, Tab, C, Val). db_erase_tab(Tab) -> db_erase_tab(val({Tab, storage_type}), Tab). db_erase_tab(ram_copies, Tab) -> ?ets_delete_table(Tab); db_erase_tab(disc_copies, Tab) -> ?ets_delete_table(Tab); -db_erase_tab(disc_only_copies, _Tab) -> ignore. +db_erase_tab(disc_only_copies, _Tab) -> ignore; +db_erase_tab({ext, _Alias, _Mod}, _Tab) -> ignore. %% assuming that Tab is a valid ets-table dets_to_ets(Tabname, Tab, File, Type, Rep, Lock) -> -- cgit v1.2.3