mnesia_ext: Add create_external and increase protocol version to monitor

new protocol version to handle new schema fields

2 files changed, 78 insertions, 13 deletions

diff --git a/lib/mnesia/src/mnesia_monitor.erl b/lib/mnesia/src/mnesia_monitor.erl
index 081d746257..66a5719276 100644
--- a/lib/mnesia/src/mnesia_monitor.erl
+++ b/lib/mnesia/src/mnesia_monitor.erl
@@ -32,6 +32,7 @@
 	 init/0,
 	 mktab/2,
 	 unsafe_mktab/2,
+         unsafe_create_external/4,
 	 mnesia_down/2,
 	 needs_protocol_conversion/1,
 	 negotiate_protocol/1,
@@ -82,9 +83,9 @@
 		going_down = [], tm_started = false, early_connects = [],
 		connecting, mq = [], remote_node_status = []}).
 
--define(current_protocol_version,  {8,2}).
+-define(current_protocol_version,  {8,3}).
 
--define(previous_protocol_version, {8,1}).
+-define(previous_protocol_version, {8,2}).
 
 start() ->
     gen_server:start_link({local, ?MODULE}, ?MODULE,
@@ -129,6 +130,8 @@ close_log(Name) ->
 unsafe_close_log(Name) ->
     unsafe_call({unsafe_close_log, Name}).
 
+unsafe_create_external(Tab, Alias, Mod, Cs) ->
+    unsafe_call({unsafe_create_external, Tab, Alias, Mod, Cs}).
 
 disconnect(Node) ->
     cast({disconnect, Node}).
@@ -193,7 +196,7 @@ protocol_version() ->
 %% A sorted list of acceptable protocols the
 %% preferred protocols are first in the list
 acceptable_protocol_versions() ->
-    [protocol_version(), ?previous_protocol_version].
+    [protocol_version(), ?previous_protocol_version, {8,1}].
 
 needs_protocol_conversion(Node) ->
     case {?catch_val({protocol, Node}), protocol_version()} of
@@ -405,6 +408,14 @@ handle_call({unsafe_close_log, Name}, _From, State) ->
     _ = disk_log:close(Name),
     {reply, ok, State};
 
+handle_call({unsafe_create_external, Tab, Alias, Mod, Cs}, _From, State) ->
+    case catch Mod:create_table(Alias, Tab, mnesia_schema:cs2list(Cs)) of
+	{'EXIT', ExitReason} ->
+	    {reply, {error, ExitReason}, State};
+	Reply ->
+	    {reply, Reply, State}
+    end;
+
 handle_call({negotiate_protocol, Mon, _Version, _Protocols}, _From, State)
   when State#state.tm_started == false ->
     State2 =  State#state{early_connects = [node(Mon) | State#state.early_connects]},
@@ -658,6 +669,7 @@ get_env(E) ->
 env() ->
     [
      access_module,
+     allow_index_on_key,
      auto_repair,
      backup_module,
      debug,
@@ -671,19 +683,23 @@ env() ->
      extra_db_nodes,
      ignore_fallback_at_startup,
      fallback_error_function,
+     fold_chunk_size,
      max_wait_for_decision,
      schema_location,
      core_dir,
      pid_sort_order,
      no_table_loaders,
      dc_dump_limit,
-     send_compressed
+     send_compressed,
+     schema
     ].
 
 default_env(access_module) ->
     mnesia;
 default_env(auto_repair) ->
     true;
+default_env(allow_index_on_key) ->
+    false;
 default_env(backup_module) ->
     mnesia_backup;
 default_env(debug) ->
@@ -709,6 +725,8 @@ default_env(ignore_fallback_at_startup) ->
     false;
 default_env(fallback_error_function) ->
     {mnesia, lkill};
+default_env(fold_chunk_size) ->
+    100;
 default_env(max_wait_for_decision) ->
     infinity;
 default_env(schema_location) ->
@@ -722,7 +740,9 @@ default_env(no_table_loaders) ->
 default_env(dc_dump_limit) ->
     4;
 default_env(send_compressed) ->
-    0.
+    0;
+default_env(schema) ->
+    [].
 
 check_type(Env, Val) ->
     try do_check_type(Env, Val)
@@ -730,6 +750,7 @@ check_type(Env, Val) ->
     end.
 
 do_check_type(access_module, A) when is_atom(A) -> A;
+do_check_type(allow_index_on_key, B) -> bool(B);
 do_check_type(auto_repair, B) -> bool(B);
 do_check_type(backup_module, B) when is_atom(B) -> B;
 do_check_type(debug, debug) -> debug;
@@ -753,6 +774,8 @@ do_check_type(extra_db_nodes, L) when is_list(L) ->
 	     (A) when is_atom(A) -> true
 	  end,
     lists:filter(Fun, L);
+do_check_type(fold_chunk_size, I) when is_integer(I), I > 0;
+				       I =:= infinity -> I;
 do_check_type(max_wait_for_decision, infinity) -> infinity;
 do_check_type(max_wait_for_decision, I) when is_integer(I), I > 0 -> I;
 do_check_type(schema_location, M) -> media(M);
@@ -766,7 +789,8 @@ do_check_type(pid_sort_order, "standard") -> standard;
 do_check_type(pid_sort_order, _) -> false;
 do_check_type(no_table_loaders, N) when is_integer(N), N > 0 -> N;
 do_check_type(dc_dump_limit,N) when is_number(N), N > 0 -> N;
-do_check_type(send_compressed, L) when is_integer(L), L >= 0, L =< 9 -> L.
+do_check_type(send_compressed, L) when is_integer(L), L >= 0, L =< 9 -> L;
+do_check_type(schema, L) when is_list(L) -> L.
 
 bool(true) -> true;
 bool(false) -> false.
diff --git a/lib/mnesia/src/mnesia_schema.erl b/lib/mnesia/src/mnesia_schema.erl
index 782493fb4f..76a9dbf26c 100644
--- a/lib/mnesia/src/mnesia_schema.erl
+++ b/lib/mnesia/src/mnesia_schema.erl
@@ -647,7 +647,15 @@ cs2list(Cs) when is_record(Cs, cstruct) ->
 cs2list(CreateList) when is_list(CreateList) ->
     CreateList;
 
-%% since 4.6
+cs2list(Cs) when element(1, Cs) == cstruct, tuple_size(Cs) == 20 ->
+    Tags = [name,type,
+	    ram_copies,disc_copies,disc_only_copies,external_copies,
+	    load_order,access_mode,majority,index,snmp,local_content,
+	    record_name,attributes,
+	    user_properties,frag_properties,storage_properties,
+	    cookie,version],
+    rec2list(Tags, Tags, 2, Cs);
+%% since vsn-4.6 (protocol 8.2 or older)
 cs2list(Cs) when element(1, Cs) == cstruct, tuple_size(Cs) == 19 ->
     Tags = [name,type,ram_copies,disc_copies,disc_only_copies,
 	    load_order,access_mode,majority,index,snmp,local_content,
@@ -657,7 +665,17 @@ cs2list(Cs) when element(1, Cs) == cstruct, tuple_size(Cs) == 19 ->
     rec2list(Tags, Tags, 2, Cs).
 
 cs2list(false, Cs) ->
-    cs2list(Cs).
+    cs2list(Cs);
+cs2list({8,3}, Cs) ->
+    cs2list(Cs);
+cs2list({8,Minor}, Cs) when Minor =:= 2; Minor =:= 1 ->
+    Orig = record_info(fields, cstruct),
+    Tags = [name,type,ram_copies,disc_copies,disc_only_copies,
+	    load_order,access_mode,majority,index,snmp,local_content,
+	    record_name,attributes,
+	    user_properties,frag_properties,storage_properties,
+	    cookie,version],
+    rec2list(Tags, Orig, 2, Cs).
 
 rec2list([Tag | Tags], [Tag | Orig], Pos, Rec) ->
     Val = element(Pos, Rec),
@@ -667,8 +685,19 @@ rec2list([], _, _Pos, _Rec) ->
 rec2list(Tags, [_|Orig], Pos, Rec) ->
     rec2list(Tags, Orig, Pos+1, Rec).
 
-normalize_cs(Cstructs, _Node) ->
-    Cstructs.
+normalize_cs(Cstructs, Node) ->
+    %% backward-compatibility hack; normalize before returning
+    case need_old_cstructs([Node]) of
+	false ->
+	    Cstructs;
+	Version ->
+	    %% some other format
+	    [convert_cs(Version, Cs) || Cs <- Cstructs]
+    end.
+
+convert_cs(Version, Cs) ->
+    Fields = [Value || {_, Value} <- cs2list(Version, Cs)],
+    list_to_tuple([cstruct|Fields]).
 
 list2cs(List) when is_list(List) ->
     Name = pick(unknown, name, List, must),
@@ -2795,11 +2824,23 @@ do_merge_schema(LockTabs0) ->
     end.
 
 fetch_cstructs(Node) ->
-    rpc:call(Node, mnesia_controller, get_remote_cstructs, []).
+    Convert = mnesia_monitor:needs_protocol_conversion(Node),
+    case rpc:call(Node, mnesia_controller, get_remote_cstructs, [])  of
+	{cstructs, Cs0, RemoteRunning1} when Convert ->
+	    {cstructs, [list2cs(cs2list(Cs)) || Cs <- Cs0], RemoteRunning1};
+	Result ->
+	    Result
+    end.
 
-need_old_cstructs() -> false.
+need_old_cstructs() ->
+    need_old_cstructs(val({schema, where_to_write})).
 
-need_old_cstructs(_Nodes) -> false.
+need_old_cstructs(Nodes) ->
+    Filter = fun(Node) -> mnesia_monitor:needs_protocol_conversion(Node) end,
+    case lists:filter(Filter, Nodes) of
+	[] -> false;
+	Ns -> lists:min([element(1, ?catch_val({protocol, Node})) || Node <- Ns])
+    end.
 
 tab_to_nodes(Tab) when is_atom(Tab) ->
     Cs = val({Tab, cstruct}),