mnesia_schema.erl

来自「OTP是开放电信平台的简称」· ERL 代码 · 共 2,011 行 · 第 1/5 页

merge_with_previous(write_property, Prop, Prev) ->
    Key = element(1, Prop),
    Prev1 = lists:keydelete(Key, 1, Prev),
    lists:sort([Prop|Prev1]);
merge_with_previous(delete_property, PropKey, Prev) ->
    lists:keydelete(PropKey, 1, Prev).

delete_table_property(Tab, PropKey) ->
    schema_transaction(fun() -> do_delete_table_property(Tab, PropKey) end).

do_delete_table_property(Tab, PropKey) ->
    TidTs = get_tid_ts_and_lock(schema, write),
    {_, _, Ts} = TidTs,
    Store = Ts#tidstore.store,
    case change_prop_in_existing_op(Tab, PropKey, delete_property, Store) of
	true ->
	    dbg_out("change_prop_in_existing_op"
		    "(~p,~p,delete_property,Store) -> true~n",
		    [Tab,PropKey]),
	    %% we have merged the table prop into the create_table op
	    ok;
	false ->
	    dbg_out("change_prop_in_existing_op"
		    "(~p,~p,delete_property,Store) -> false~n",
		    [Tab,PropKey]),
	    %% this must be an existing table
	    get_tid_ts_and_lock(Tab, none),
	    insert_schema_ops(TidTs, 
			      make_delete_table_properties(Tab, [PropKey]))
    end.

make_delete_table_properties(Tab, PropKeys) ->
    ensure_writable(schema),
    Cs = incr_version(val({Tab, cstruct})),
    ensure_active(Cs),
    make_delete_table_properties(Tab, PropKeys, Cs).

make_delete_table_properties(Tab, [PropKey | PropKeys], Cs) ->
    OldProps = Cs#cstruct.user_properties,
    Props = lists:keydelete(PropKey, 1, OldProps),
    Cs2 = Cs#cstruct{user_properties = Props},
    verify_cstruct(Cs2),
    [{op, delete_property, cs2list(Cs2), PropKey} |
     make_delete_table_properties(Tab, PropKeys, Cs2)];
make_delete_table_properties(_Tab, [], _Cs) ->
    [].

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

%% Ensure that the transaction can be committed even 
%% if the node crashes and Mnesia is restarted
prepare_commit(Tid, Commit, WaitFor) ->
    case Commit#commit.schema_ops of
	[] ->
	    {false, Commit, optional};
	OrigOps ->
	    {Modified, Ops, DumperMode} = 
		prepare_ops(Tid, OrigOps, WaitFor, false, [], optional),
	    InitBy = schema_prepare,
	    GoodRes = {Modified, 
		       Commit#commit{schema_ops = lists:reverse(Ops)},
		       DumperMode},
	    case DumperMode of
		optional ->
		    dbg_out("Transaction log dump skipped (~p): ~w~n",
			    [DumperMode, InitBy]);
		mandatory ->
		    case mnesia_controller:sync_dump_log(InitBy) of
			dumped ->
			    GoodRes;
			{error, Reason} ->
			    mnesia:abort(Reason)
		    end
	    end,
	    case Ops of
		[] -> 
		    ignore;
		_ ->
		    %% We need to grab a dumper lock here, the log may not
		    %% be dumped by others, during the schema commit phase.
		    mnesia_controller:wait_for_schema_commit_lock()
	    end,
	    GoodRes
    end.

prepare_ops(Tid, [Op | Ops], WaitFor, Changed, Acc, DumperMode) ->
    case prepare_op(Tid, Op, WaitFor) of
        {true, mandatory} -> 
	    prepare_ops(Tid, Ops, WaitFor, Changed, [Op | Acc], mandatory);
        {true, optional} -> 
	    prepare_ops(Tid, Ops, WaitFor, Changed, [Op | Acc], DumperMode);
        {true, Ops2, mandatory} -> 
	    prepare_ops(Tid, Ops, WaitFor, true, Ops2 ++ Acc, mandatory);
        {true, Ops2, optional} -> 
	    prepare_ops(Tid, Ops, WaitFor, true, Ops2 ++ Acc, DumperMode);
	{false, optional} -> 
	    prepare_ops(Tid, Ops, WaitFor, true, Acc, DumperMode)
    end;
prepare_ops(_Tid, [], _WaitFor, Changed, Acc, DumperMode) ->
    {Changed, Acc, DumperMode}.
    
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% Prepare for commit
%% returns true if Op should be included, i.e. unmodified
%%         {true, Operation} if NewRecs should be included, i.e. modified
%%         false if Op should NOT be included, i.e. modified
%%
prepare_op(_Tid, {op, rec, unknown, Rec}, _WaitFor) ->
    {{Tab, Key}, Items, _Op} = Rec,
    case val({Tab, storage_type}) of
        unknown ->
            {false, optional};
        Storage ->
            mnesia_tm:prepare_snmp(Tab, Key, Items), % May exit
            {true, [{op, rec, Storage, Rec}], optional}
    end;

prepare_op(_Tid, {op, announce_im_running, Node, SchemaDef, Running, RemoteRunning}, _WaitFor) ->
    SchemaCs = list2cs(SchemaDef),
    if 	
	Node == node() -> %% Announce has already run on local node 
	    ignore;       %% from do_merge_schema
	true ->
	    NewNodes = mnesia_lib:uniq(Running++RemoteRunning) -- val({current,db_nodes}),
	    mnesia_lib:set(prepare_op, {announce_im_running,NewNodes}),
	    announce_im_running(NewNodes, SchemaCs)
    end,
    {false, optional};

prepare_op(_Tid, {op, sync_trans}, {part, CoordPid}) ->    
    CoordPid ! {sync_trans, self()},
    receive 
	{sync_trans, CoordPid} ->
	    {false, optional};
	Else ->
	    mnesia_lib:verbose("sync_op terminated due to ~p~n", [Else]),
	    mnesia:abort(Else)
    end;

prepare_op(_Tid, {op, sync_trans}, {coord, Nodes}) ->    
    case receive_sync(Nodes, []) of
	{abort, Reason} ->
	    mnesia_lib:verbose("sync_op terminated due to ~p~n", [Reason]),
	    mnesia:abort(Reason);
	Pids ->
	    [Pid ! {sync_trans, self()} || Pid <- Pids],
	    {false, optional}
    end;
prepare_op(Tid, {op, create_table, TabDef}, _WaitFor) ->
    Cs = list2cs(TabDef),
    Storage = mnesia_lib:cs_to_storage_type(node(), Cs),
    UseDir = mnesia_monitor:use_dir(),
    Tab = Cs#cstruct.name,
    case Storage of
        disc_copies when UseDir == false ->
	    UseDirReason = {bad_type, Tab, Storage, node()},
            mnesia:abort(UseDirReason);
        disc_only_copies when UseDir == false ->
	    UseDirReason = {bad_type, Tab, Storage, node()},
            mnesia:abort(UseDirReason);
	ram_copies ->
	    mnesia_lib:set({Tab, create_table},true),
	    create_ram_table(Tab, Cs#cstruct.type),
	    insert_cstruct(Tid, Cs, false),
	    {true, optional};
	disc_copies ->
	    mnesia_lib:set({Tab, create_table},true),
	    create_ram_table(Tab, Cs#cstruct.type),
	    create_disc_table(Tab),
	    insert_cstruct(Tid, Cs, false),
	    {true, optional};	
	disc_only_copies ->
	    mnesia_lib:set({Tab, create_table},true),
	    create_disc_only_table(Tab,Cs#cstruct.type),
	    insert_cstruct(Tid, Cs, false),
	    {true, optional};
        unknown -> %% No replica on this node
	    mnesia_lib:set({Tab, create_table},true),
	    insert_cstruct(Tid, Cs, false),
            {true, optional}
    end;

prepare_op(Tid, {op, add_table_copy, Storage, Node, TabDef}, _WaitFor) ->
    Cs = list2cs(TabDef),
    Tab = Cs#cstruct.name,

    if
	Tab == schema ->
	    {true, optional};
	
	Node == node() ->
	    case mnesia_lib:val({schema, storage_type}) of 
		ram_copies when Storage /= ram_copies -> 
		    Error = {combine_error, Tab, "has no disc", Node},
		    mnesia:abort(Error);
		_  ->
		    ok
	    end,   
	    %% Tables are created by mnesia_loader get_network code
	    insert_cstruct(Tid, Cs, true),
	    case mnesia_controller:get_network_copy(Tab, Cs) of
		{loaded, ok} ->
		    {true, optional};
		{not_loaded, ErrReason} ->
		    Reason = {system_limit, Tab, {Node, ErrReason}},
		    mnesia:abort(Reason)
	    end;
	Node /= node() ->
	    %% Verify that ram table not has been dumped to disc
	    if
		Storage /= ram_copies ->
		    case mnesia_lib:schema_cs_to_storage_type(node(), Cs) of
			ram_copies ->
			    Dat = mnesia_lib:tab2dcd(Tab),
			    case mnesia_lib:exists(Dat) of
				true ->
				    mnesia:abort({combine_error, Tab, Storage,
						  "Table dumped to disc", node()});
				false ->
				    ok
			    end;
			_ ->
			    ok
		    end;
		true ->
		    ok
	    end,
	    insert_cstruct(Tid, Cs, true),
	    {true, optional}
    end;

prepare_op(Tid, {op, del_table_copy, _Storage, Node, TabDef}, _WaitFor) ->
    Cs = list2cs(TabDef),
    Tab = Cs#cstruct.name,
    
    if
	%% Schema table lock is always required to run a schema op.
	%% No need to look it.
	node(Tid#tid.pid) == node(), Tab /= schema -> 
	    Self = self(),
	    Pid = spawn_link(fun() -> lock_del_table(Tab, Node, Cs, Self) end),
	    put(mnesia_lock, Pid),
	    receive 
		{Pid, updated} -> 
		    {true, optional};
		{Pid, FailReason} ->
		    mnesia:abort(FailReason);
		{'EXIT', Pid, Reason} ->
		    mnesia:abort(Reason)
	    end;	
	true ->
	    {true, optional}
    end;

prepare_op(_Tid, {op, change_table_copy_type,  N, FromS, ToS, TabDef}, _WaitFor)
  when N == node() ->
    Cs = list2cs(TabDef),
    Tab = Cs#cstruct.name,

    NotActive = mnesia_lib:not_active_here(Tab),
    
    if 
	NotActive == true ->
	    mnesia:abort({not_active, Tab, node()});
	
	Tab == schema ->	    
	    case {FromS, ToS} of
		{ram_copies, disc_copies} ->
		    case mnesia:system_info(schema_location) of
			opt_disc ->
			    ignore;
			_ ->
			    mnesia:abort({combine_error,  Tab, node(),
					  "schema_location must be opt_disc"})
		    end,
		    Dir = mnesia_lib:dir(),    
		    case opt_create_dir(true, Dir) of
			ok ->
			    purge_dir(Dir, []),
			    mnesia_log:purge_all_logs(),
			    set(use_dir, true),
			    mnesia_log:init(),
			    Ns = val({current, db_nodes}), %mnesia_lib:running_nodes(),
			    F = fun(U) -> mnesia_recover:log_mnesia_up(U) end,
			    lists:foreach(F, Ns),

			    mnesia_dumper:raw_named_dump_table(Tab, dmp),
			    mnesia_checkpoint:tm_change_table_copy_type(Tab, FromS, ToS);
			{error, Reason} ->
			    mnesia:abort(Reason)
		    end;
		{disc_copies, ram_copies} ->
		    Ltabs = val({schema, local_tables}) -- [schema],
		    Dtabs = [L || L <- Ltabs,
				  val({L, storage_type}) /= ram_copies],
		    verify([], Dtabs, {"Disc resident tables", Dtabs, N});
		_ ->
		    mnesia:abort({combine_error, Tab, ToS})
	    end;
	
	FromS == ram_copies ->	    
	    case mnesia_monitor:use_dir() of
		true -> 
		    Dat = mnesia_lib:tab2dcd(Tab),
		    case mnesia_lib:exists(Dat) of
			true ->
			    mnesia:abort({combine_error, Tab, node(),
					  "Table dump exists"});
			false ->
			    case ToS of 
				disc_copies -> 
				    mnesia_log:ets2dcd(Tab, dmp);
				disc_only_copies ->
				    mnesia_dumper:raw_named_dump_table(Tab, dmp)
			    end,
			    mnesia_checkpoint:tm_change_table_copy_type(Tab, FromS, ToS)
		    end;
		false ->
		    mnesia:abort({has_no_disc, node()})
	    end;
	
	FromS == disc_copies, ToS == disc_only_copies ->
	    mnesia_dumper:raw_named_dump_table(Tab, dmp);
	FromS == disc_only_copies ->
	    Type = Cs#cstruct.type,
	    create_ram_table(Tab, Type),
	    Datname = mnesia_lib:tab2dat(Tab),
	    Repair = mnesia_monitor:get_env(auto_repair),
	    case mnesia_lib:dets_to_ets(Tab, Tab, Datname, Type, Repair, no) of
		loaded -> ok;
		Reason ->
		    Err = "Failed to copy disc data to ram",
		    mnesia:abort({system_limit, Tab, {Err,Reason}})
	    end;
	true ->
	    ignore
    end,
    {true, mandatory};

prepare_op(_Tid, {op, change_table_copy_type,  N, _FromS, _ToS, _TabDef}, _WaitFor)
  when N /= node() ->
    {true, mandatory};

prepare_op(_Tid, {op, delete_table, _TabDef}, _WaitFor) ->
    {true, mandatory};

prepare_op(_Tid, {op, dump_table, unknown, TabDef}, _WaitFor) ->
    Cs = list2cs(TabDef),
    Tab = Cs#cstruct.name,
    case lists:member(node(), Cs#cstruct.ram_copies) of
        true ->
	    case mnesia_monitor:use_dir() of
		true -> 
		    mnesia_log:ets2dcd(Tab, dmp),
		    Size = mnesia:table_info(Tab, size),
		    {true, [{op, dump_table, Size, TabDef}], optional};
		false ->
		    mnesia:abort({has_no_disc, node()})
	    end;
        false ->
            {false, optional}
    end;

prepare_op(_Tid, {op, add_snmp, Ustruct, TabDef}, _WaitFor) ->
    Cs = list2cs(TabDef),
    case mnesia_lib:cs_to_storage_type(node(), Cs) of
        unknown ->
            {true, optional};
        Storage ->
            Tab = Cs#cstruct.name,
            Stab = mnesia_snmp_hook:create_table(Ustruct, Tab, Storage),
            mnesia_lib:set({Tab, {index, snmp}}, Stab),
            {true, optional}
    end;

prepare_op(_Tid, {op, transform, ignore, _TabDef}, _WaitFor) ->
    {true, mandatory};   %% Apply schema changes only.
prepare_op(_Tid, {op, transform, Fun, TabDef}, _WaitFor) ->
    Cs = list2cs(TabDef),
    case mnesia_lib:cs_to_storage_type(node(), Cs) of
        unknown ->
            {true, mandatory};
        Storage ->
            Tab = Cs#cstruct.name,
            RecName = Cs#cstruct.record_name,
	    Type = Cs#cstruct.type,
            NewArity = length(Cs#cstruct.attributes) + 1,
	    mnesia_lib:db_fixtable(Storage, Tab, true),
            Key = mnesia_lib:db_first(Tab),
	    Op = {op, transform, Fun, TabDef},
            case catch transform_objs(Fun, Tab, RecName, 
				      Key, NewArity, Storage, Type, [Op]) of
                {'EXIT', Reason} ->
		    mnesia_lib:db_fixtable(Storage, Tab, false),
                    exit({"Bad transform function", Tab, Fun, node(), Reason});
                Objs ->
		    mnesia_lib:db_fixtable(Storage, Tab, false),