mnesia_ext: Implement ext copies index

Make ram_copies index always use ordered_set

And use index type as prefered type not a implementation requirement,
the standard implmentation will currently ignore the prefered type.

1 files changed, 308 insertions, 126 deletions

diff --git a/lib/mnesia/src/mnesia_index.erl b/lib/mnesia/src/mnesia_index.erl
index 0c882c0df6..902d0c12ae 100644
--- a/lib/mnesia/src/mnesia_index.erl
+++ b/lib/mnesia/src/mnesia_index.erl
@@ -23,8 +23,8 @@
 
 -module(mnesia_index).
 -export([read/5,
-	 add_index/5,
-	 delete_index/3,
+	 add_index/6,
+	 delete_index/4,
 	 del_object_index/5,
 	 clear_index/4,
 	 dirty_match_object/3,
@@ -39,19 +39,21 @@
 	 get_index_table/3,
 	 
 	 tab2filename/2,
-	 tab2tmp_filename/2,
 	 init_index/2,
 	 init_indecies/3,
 	 del_transient/2,
 	 del_transient/3,
-	 del_index_table/3]).
+	 del_index_table/3,
+
+         index_info/2,
+	 ext_index_instances/1]).
 
 -import(mnesia_lib, [val/1, verbose/2]).
 -include("mnesia.hrl").
 
 -record(index, {setorbag, pos_list}).
 
-%% read an object list throuh its index table
+%% read an object list through its index table
 %% we assume that table Tab has index on attribute number Pos
 
 read(Tid, Store, Tab, IxKey, Pos) ->
@@ -64,69 +66,103 @@ read(Tid, Store, Tab, IxKey, Pos) ->
 	    ResList
     end.
 
-add_index(Index, Tab, Key, Obj, Old) ->    
-    add_index2(Index#index.pos_list, Index#index.setorbag, Tab, Key, Obj, Old).
-
-add_index2([{Pos, Ixt} |Tail], bag, Tab, K, Obj, OldRecs) ->
-    db_put(Ixt, {element(Pos, Obj), K}),
-    add_index2(Tail, bag, Tab, K, Obj, OldRecs);
-add_index2([{Pos, Ixt} |Tail], Type, Tab, K, Obj, OldRecs0) ->
+ext_index_instances(Tab) ->
+    #index{pos_list = PosL} = val({Tab, index_info}),
+    lists:foldr(
+      fun({_, {{ext,Alias,Mod}, Tag}}, Acc) ->
+	      [{Alias, Mod, Tag}|Acc];
+	 (_, Acc) ->
+	      Acc
+      end, [], PosL).
+
+
+add_index(#index{pos_list = PosL, setorbag = SorB},
+	  Storage, Tab, Key, Obj, Old) ->
+    add_index2(PosL, SorB, Storage, Tab, Key, Obj, Old).
+
+add_index2([{{Pos,Type}, Ixt} |Tail], bag, Storage, Tab, K, Obj, OldRecs) ->
+    ValsF = index_vals_f(Storage, Tab, Pos),
+    Vals = ValsF(Obj),
+    put_index_vals(Type, Ixt, Vals, K),
+    add_index2(Tail, bag, Storage, Tab, K, Obj, OldRecs);
+add_index2([{{Pos, Type}, Ixt} |Tail], SorB, Storage, Tab, K, Obj, OldRecs0) ->
     %% Remove old tuples in index if Tab is updated
+    ValsF = index_vals_f(Storage, Tab, Pos),    NewVals = ValsF(Obj),
     OldRecs1 = case OldRecs0 of
-		   undefined -> mnesia_lib:db_get(Tab, K);
+		   undefined -> mnesia_lib:db_get(Storage, Tab, K);
 		   _ -> OldRecs0
 	       end,
-    IdxVal = element(Pos, Obj),
-    case [Old || Old <- OldRecs1, element(Pos, Old) =/= IdxVal] of
-	[] when OldRecs1 =:= [] ->  %% Write
-	    db_put(Ixt, {element(Pos, Obj), K}),
-	    add_index2(Tail, Type, Tab, K, Obj, OldRecs0);
+    IdxVal = ValsF(Obj),
+    case [Old || Old <- OldRecs1, ValsF(Old) =/= IdxVal] of
+	[] when OldRecs1 =:= [] -> % Write
+	    put_index_vals(Type, Ixt, NewVals, K),
+	    add_index2(Tail, SorB, Storage, Tab, K, Obj, OldRecs1);
 	[] -> %% when OldRecs1 =/= [] Update without modifying index field
-	    add_index2(Tail, Type, Tab, K, Obj, OldRecs0);
+	    add_index2(Tail, SorB, Storage, Tab, K, Obj, OldRecs1);
 	OldRecs -> %% Update
-	    db_put(Ixt, {element(Pos, Obj), K}),
-	    del_ixes(Ixt, OldRecs, Pos, K),
-	    add_index2(Tail, Type, Tab, K, Obj, OldRecs0)
+	    put_index_vals(Type, Ixt, NewVals, K),
+	    [del_ixes(Type, Ixt, ValsF, OldObj, K) || OldObj <- OldRecs],
+	    add_index2(Tail, SorB, Storage, Tab, K, Obj, OldRecs1)
     end;
-add_index2([], _, _Tab, _K, _Obj, _) -> ok.
+add_index2([], _, _, _Tab, _K, _Obj, _) -> ok.
+
+delete_index(Index, Storage, Tab, K) ->
+    delete_index2(Index#index.pos_list, Storage, Tab, K).
 
-delete_index(Index, Tab, K) ->
-    delete_index2(Index#index.pos_list, Tab, K).
+delete_index2([{{Pos, Type}, Ixt} | Tail], Storage, Tab, K) ->
+    DelObjs = mnesia_lib:db_get(Storage, Tab, K),
+    ValsF = index_vals_f(Storage, Tab, Pos),
+    [del_ixes(Type, Ixt, ValsF, Obj, K) || Obj <- DelObjs],
+    delete_index2(Tail, Storage, Tab, K);
+delete_index2([], _Storage, _Tab, _K) -> ok.
 
-delete_index2([{Pos, Ixt} | Tail], Tab, K) ->
-    DelObjs = mnesia_lib:db_get(Tab, K), 
-    del_ixes(Ixt, DelObjs, Pos, K),
-    delete_index2(Tail, Tab, K);
-delete_index2([], _Tab, _K) -> ok.
+put_index_vals(ordered, Ixt, Vals, K) ->
+    [db_put(Ixt, {{V, K}}) || V <- Vals];
+put_index_vals(bag, Ixt, Vals, K) ->
+    [db_put(Ixt, {V, K}) || V <- Vals].
 
 
-del_ixes(_Ixt, [], _Pos, _L) -> ok;
-del_ixes(Ixt, [Obj | Tail], Pos, Key) ->
-    db_match_erase(Ixt, {element(Pos, Obj), Key}),
-    del_ixes(Ixt, Tail, Pos, Key).
+del_ixes(bag, Ixt, ValsF, Obj, Key) ->
+    Vals = ValsF(Obj),
+    [db_match_erase(Ixt, {V, Key}) || V <- Vals];
+del_ixes(ordered, Ixt, ValsF, Obj, Key) ->
+    Vals = ValsF(Obj),
+    [db_erase(Ixt, {V,Key}) || V <- Vals].
 
-del_object_index(Index, Tab, K, Obj, Old) ->
-    del_object_index2(Index#index.pos_list, Index#index.setorbag, Tab, K, Obj, Old).
+del_object_index(#index{pos_list = PosL, setorbag = SorB}, Storage, Tab, K, Obj) ->
+    del_object_index2(PosL, SorB, Storage, Tab, K, Obj).
 
-del_object_index2([], _, _Tab, _K, _Obj, _Old) -> ok;
-del_object_index2([{Pos, Ixt} | Tail], SoB, Tab, K, Obj, Old) ->
+del_object_index2([], _, _Storage, _Tab, _K, _Obj) -> ok;
+del_object_index2([{{Pos, Type}, Ixt} | Tail], SoB, Storage, Tab, K, Obj) ->
+    ValsF = index_vals_f(Storage, Tab, Pos),
     case SoB of
 	bag -> 
-	    del_object_bag(Tab, K, Obj, Pos, Ixt, Old);
+	    del_object_bag(Type, ValsF, Tab, K, Obj, Ixt);
 	_ -> %% If set remove the tuple in index table
-	    del_ixes(Ixt, [Obj], Pos, K)	
+	    del_ixes(Type, Ixt, ValsF, Obj, K)
     end,
-    del_object_index2(Tail, SoB, Tab, K, Obj, Old).
-
-del_object_bag(Tab, Key, Obj, Pos, Ixt, undefined) -> 
-    IxKey = element(Pos, Obj),
-    Old = [X || X <-  mnesia_lib:db_get(Tab, Key), element(Pos, X) =:= IxKey],
-    del_object_bag(Tab, Key, Obj, Pos, Ixt, Old);
-%% If Tab type is bag we need remove index identifier if the object being
-%% deleted was the last one
-del_object_bag(_Tab, Key, Obj, Pos, Ixt, Old) when Old =:= [Obj] ->
-    del_ixes(Ixt, [Obj], Pos, Key);
-del_object_bag(_Tab, _Key, _Obj, _Pos, _Ixt, _Old) -> ok.
+    del_object_index2(Tail, SoB, Storage, Tab, K, Obj).
+
+del_object_bag(Type, ValsF, Tab, Key, Obj, Ixt) ->
+    IxKeys = ValsF(Obj),
+    Found = [{X, ValsF(X)} || X <- mnesia_lib:db_get(Tab, Key)],
+    del_object_bag_(IxKeys, Found, Type, Tab, Key, Obj, Ixt).
+
+del_object_bag_([IxK|IxKs], Found, Type, Tab, Key, Obj, Ixt) ->
+    case [X || {X, Ixes} <- Found, lists:member(IxK, Ixes)] of
+        [Old] when Old =:= Obj ->
+	    case Type of
+		bag ->
+                    db_match_erase(Ixt, {IxK, Key});
+		ordered ->
+		    db_erase(Ixt, {{IxK, Key}})
+	    end;
+        _ ->
+	    ok
+    end,
+    del_object_bag_(IxKs, Found, Type, Tab, Key, Obj, Ixt);
+del_object_bag_([], _, _, _, _, _, _) ->
+    ok.
 
 clear_index(Index, Tab, K, Obj) ->
     clear_index2(Index#index.pos_list, Tab, K, Obj).
@@ -138,8 +174,9 @@ clear_index2([{_Pos, Ixt} | Tail], Tab, K, Obj) ->
 
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
 
-dirty_match_object(Tab, Pat, Pos) ->
+dirty_match_object(Tab, Pat, Pos) when is_integer(Pos) ->
     %% Assume that we are on the node where the replica is
+    %% Cannot use index plugins here, as they don't map to match patterns
     case element(2, Pat) of
 	'_' ->
 	    IxKey = element(Pos, Pat),
@@ -161,7 +198,7 @@ realkeys(Tab, Pos, IxKey) ->
     Index = get_index_table(Tab, Pos),
     db_get(Index, IxKey). % a list on the form [{IxKey, RealKey1} , ....
     
-dirty_select(Tab, Spec, Pos) ->
+dirty_select(Tab, Spec, Pos) when is_integer(Pos) ->
     %% Assume that we are on the node where the replica is
     %% Returns the records without applying the match spec
     %% The actual filtering is handled by the caller
@@ -171,59 +208,80 @@ dirty_select(Tab, Spec, Pos) ->
     lists:append([mnesia_lib:db_get(StorageType, Tab, Key) || {_,Key} <- RealKeys]).
 
 dirty_read(Tab, IxKey, Pos) ->
-    ResList = mnesia:dirty_rpc(Tab, ?MODULE, dirty_read2,
-			       [Tab, IxKey, Pos]),
-    case val({Tab, setorbag}) of
-	bag ->
-	    %% Remove all tuples which don't include Ixkey
-	    mnesia_lib:key_search_all(IxKey, Pos, ResList);
-	_ -> 
-	    ResList
-    end.
+    mnesia:dirty_rpc(Tab, ?MODULE, dirty_read2,
+		     [Tab, IxKey, Pos]).
 
 dirty_read2(Tab, IxKey, Pos) ->
-    Ix = get_index_table(Tab, Pos),
-    Keys = db_match(Ix, {IxKey, '$1'}),
-    r_keys(Keys, Tab, []).
-
-r_keys([[H]|T],Tab,Ack) -> 
-    V = mnesia_lib:db_get(Tab, H),
-    r_keys(T, Tab, V ++ Ack);
-r_keys([], _, Ack) ->
-    Ack.
+    #index{pos_list = PosL} = val({Tab, index_info}),
+    Storage = val({Tab, storage_type}),
+    {Type, Ixt} = pick_index(PosL, Tab, Pos),
+    Pat = case Type of
+	      ordered -> [{{{IxKey, '$1'}}, [], ['$1']}];
+	      bag     -> [{{IxKey, '$1'}, [], ['$1']}]
+	  end,
+    Keys = db_select(Ixt, Pat),
+    ValsF = index_vals_f(Storage, Tab, Pos),
+    lists:reverse(
+      lists:foldl(
+	fun(K, Acc) ->
+		lists:foldl(
+		  fun(Obj, Acc1) ->
+			  case lists:member(IxKey, ValsF(Obj)) of
+			      true -> [Obj|Acc1];
+			      false -> Acc1
+			  end
+		  end, Acc, mnesia_lib:db_get(Storage, Tab, K))
+	end, [], Keys)).
+
+pick_index([{{{Pfx,_},IxType}, Ixt}|_], _Tab, {_} = Pfx) ->
+    {IxType, Ixt};
+pick_index([{{Pos,IxType}, Ixt}|_], _Tab, Pos) ->
+    {IxType, Ixt};
+pick_index([_|T], Tab, Pos) ->
+    pick_index(T, Tab, Pos);
+pick_index([], Tab, Pos) ->
+    mnesia:abort({no_exist, Tab, {index, Pos}}).
+
 	    
 
 %%%%%%% Creation, Init and deletion routines for index tables
 %% We can have several indexes on the same table
 %% this can be a fairly costly operation if table is *very* large
 
-tab2filename(Tab, Pos) ->
+tab2filename(Tab, {A}) when is_atom(A) ->
+    mnesia_lib:dir(Tab) ++ "_-" ++ atom_to_list(A) ++ "-.DAT";
+tab2filename(Tab, T) when is_tuple(T) ->
+    tab2filename(Tab, element(1, T));
+tab2filename(Tab, Pos) when is_integer(Pos) ->
     mnesia_lib:dir(Tab) ++ "_" ++ integer_to_list(Pos) ++ ".DAT".
 
-tab2tmp_filename(Tab, Pos) ->
-    mnesia_lib:dir(Tab) ++ "_" ++ integer_to_list(Pos) ++ ".TMP".
-        
 init_index(Tab, Storage) ->
-    PosList = val({Tab, index}),
+    Cs = val({Tab, cstruct}),
+    PosList = Cs#cstruct.index,
     init_indecies(Tab, Storage, PosList).
 
 init_indecies(Tab, Storage, PosList) ->
     case Storage of
 	unknown ->
 	    ignore;
+	{ext, Alias, Mod} ->
+	    init_ext_index(Tab, Storage, Alias, Mod, PosList);
 	disc_only_copies ->
-	    init_disc_index(Tab, PosList);
+	    init_disc_index(Tab, Storage, PosList);
 	ram_copies ->
-	    make_ram_index(Tab, PosList);
+	    make_ram_index(Tab, Storage, PosList);
 	disc_copies ->
-	    make_ram_index(Tab, PosList)
+	    make_ram_index(Tab, Storage, PosList)
     end.
 
 %% works for both ram and disc indexes
 
 del_index_table(_, unknown, _) ->
     ignore;
-del_index_table(Tab, Storage, Pos) ->
+del_index_table(Tab, Storage, {_} = Pos) ->
+    delete_transient_index(Tab, Pos, Storage),
+    mnesia_lib:del({Tab, index}, Pos);
+del_index_table(Tab, Storage, Pos) when is_integer(Pos) ->
     delete_transient_index(Tab, Pos, Storage),
     mnesia_lib:del({Tab, index}, Pos).
 
@@ -236,13 +294,24 @@ del_transient(Tab, [Pos | Tail], Storage) ->
     delete_transient_index(Tab, Pos, Storage),
     del_transient(Tab, Tail, Storage).
 
+delete_transient_index(Tab, Pos, {ext, Alias, Mod}) ->
+    PosInfo = case Pos of
+		  _ when is_integer(Pos) ->
+		      Cs = val({Tab, cstruct}),
+		      lists:keyfind(Pos, 1, Cs#cstruct.index);
+		  {P, T} -> {P, T}
+	      end,
+    Tag = {Tab, index, PosInfo},
+    Mod:close_table(Alias, Tag),
+    Mod:delete_table(Alias, Tag),
+    del_index_info(Tab, Pos),
+    mnesia_lib:unset({Tab, {index, Pos}});
 delete_transient_index(Tab, Pos, disc_only_copies) ->
     Tag = {Tab, index, Pos},
     mnesia_monitor:unsafe_close_dets(Tag),
     _ = file:delete(tab2filename(Tab, Pos)),
     del_index_info(Tab, Pos), %% Uses val(..)
     mnesia_lib:unset({Tab, {index, Pos}});
-
 delete_transient_index(Tab, Pos, _Storage) ->
     Ixt = val({Tab, {index, Pos}}),
     ?ets_delete_table(Ixt),
@@ -252,11 +321,12 @@ delete_transient_index(Tab, Pos, _Storage) ->
 %%%%% misc functions for the index create/init/delete functions above
 
 %% assuming that the file exists.
-init_disc_index(_Tab, []) ->
+init_disc_index(_Tab, _Storage, []) ->
     done;
-init_disc_index(Tab, [Pos | Tail]) when is_integer(Pos) ->
+init_disc_index(Tab, disc_only_copies, [{Pos,_Pref} | Tail]) ->
+    PosInfo = {Pos, bag},
     Fn = tab2filename(Tab, Pos),
-    IxTag = {Tab, index, Pos},
+    IxTag = {Tab, index, PosInfo},
     _ = file:delete(Fn),
     Args = [{file, Fn}, {keypos, 1}, {type, bag}],
     mnesia_monitor:open_dets(IxTag, Args),
@@ -269,16 +339,58 @@ init_disc_index(Tab, [Pos | Tail]) when is_integer(Pos) ->
     mnesia_lib:db_fixtable(Storage, Tab, true),
     ok = dets:init_table(IxTag, create_fun(Init, Tab, Pos)),
     mnesia_lib:db_fixtable(Storage, Tab, false),
-    mnesia_lib:set({Tab, {index, Pos}}, IxTag),
-    add_index_info(Tab, val({Tab, setorbag}), {Pos, {dets, IxTag}}),
-    init_disc_index(Tab, Tail).
+    mnesia_lib:set({Tab, {index, PosInfo}}, IxTag),
+    add_index_info(Tab, val({Tab, setorbag}), {PosInfo, {dets, IxTag}}),
+    init_disc_index(Tab, Storage, Tail).
+
+init_ext_index(_, _, _, _, []) ->
+    done;
+init_ext_index(Tab, Storage, Alias, Mod, [{Pos,Type} | Tail]) ->
+    PosInfo = {Pos, Type},
+    IxTag = {Tab, index, PosInfo},
+    _Res = Mod:create_table(Alias, IxTag, []),
+    CS = val({Tab, cstruct}),
+    Mod:load_table(Alias, IxTag, init_index, mnesia_schema:cs2list(CS)),
+    case Mod:is_index_consistent(Alias, IxTag) of
+        false ->
+	    Mod:index_is_consistent(Alias, IxTag, false),
+	    Mod:match_delete(Alias, IxTag, '_'),
+	    IxValsF = index_vals_f(Storage, Tab, Pos),
+            IxObjF = case Type of
+                         bag     -> fun(IxVal, Key) -> {IxVal, Key} end;
+                         ordered -> fun(IxVal, Key) -> {{IxVal, Key}} end
+                     end,
+            mnesia_lib:db_fixtable(Storage, Tab, true),
+            mnesia_lib:db_foldl(
+	      Storage,
+	      fun(Rec, Acc) ->
+		      Key = element(2, Rec),
+		      lists:foreach(
+			fun(V) ->
+				IxObj = IxObjF(V, Key),
+				Mod:insert(Alias, IxTag, IxObj)
+			end, IxValsF(Rec)),
+		      Acc
+	      end, ok, Tab,
+	      [{'_', [], ['$_']}], 100),
+	    Mod:index_is_consistent(Alias, IxTag, true);
+	true ->
+            ignore
+    end,
+
+    mnesia_lib:set({Tab, {index, PosInfo}}, IxTag),
+
+    add_index_info(Tab, val({Tab, setorbag}), {PosInfo, {Storage, IxTag}}),
+    init_ext_index(Tab, Storage, Alias, Mod, Tail).
 
 create_fun(Cont, Tab, Pos) ->
+    IxF = index_vals_f(disc_only_copies, Tab, Pos),
     fun(read) ->
 	    Data = 
 		case Cont of
 		    {start, KeysPerChunk} ->
-			mnesia_lib:db_init_chunk(disc_only_copies, Tab, KeysPerChunk);
+			mnesia_lib:db_init_chunk(
+			  disc_only_copies, Tab, KeysPerChunk);
 		    '$end_of_table' -> 
 			'$end_of_table';
 		    _Else ->
@@ -288,56 +400,82 @@ create_fun(Cont, Tab, Pos) ->
 		'$end_of_table' ->
 		    end_of_input;
 		{Recs, Next} ->
-		    IdxElems = [{element(Pos, Obj), element(2, Obj)} || Obj <- Recs],
+		    IdxElems = lists:flatmap(
+				 fun(Obj) ->
+					 PrimK = element(2, Obj),
+					 [{V, PrimK} || V <- IxF(Obj)]
+				 end, Recs),
 		    {IdxElems, create_fun(Next, Tab, Pos)}
 	    end;
        (close) ->
 	    ok
     end.
 
-make_ram_index(_, []) -> 
+make_ram_index(_, _, []) ->
     done;
-make_ram_index(Tab, [Pos | Tail]) ->
-    add_ram_index(Tab, Pos),
-    make_ram_index(Tab, Tail).
+make_ram_index(Tab, Storage, [Pos | Tail]) ->
+    add_ram_index(Tab, Storage, Pos),
+    make_ram_index(Tab, Storage, Tail).
 
-add_ram_index(Tab, Pos) when is_integer(Pos) ->
-    verbose("Creating index for ~w ~n", [Tab]),
+add_ram_index(Tab, Storage, {Pos, _Pref}) ->
+    Type = ordered,
+    verbose("Creating index for ~w ~p ~p~n", [Tab, Pos, Type]),
     SetOrBag = val({Tab, setorbag}),
-    IndexType = case SetOrBag of
-        set -> duplicate_bag;
-        ordered_set -> duplicate_bag;
-        bag -> bag
-    end,
-    Index = mnesia_monitor:mktab(mnesia_index, [IndexType, public]),
+    IxValsF = index_vals_f(Storage, Tab, Pos),
+    IxFun = fun(Val, Key) -> {{Val, Key}} end,
+    Index = mnesia_monitor:mktab(mnesia_index, [ordered_set, public]),
     Insert = fun(Rec, _Acc) ->
-		     true = ?ets_insert(Index, {element(Pos, Rec), element(2, Rec)})
+		     PrimK = element(2, Rec),
+		     true = ?ets_insert(
+			       Index, [IxFun(V, PrimK)
+				       || V <- IxValsF(Rec)])
 	     end,
     mnesia_lib:db_fixtable(ram_copies, Tab, true),
-    true = ets:foldl(Insert, true, Tab),
+    true = mnesia_lib:db_foldl(Storage, Insert, true, Tab),
     mnesia_lib:db_fixtable(ram_copies, Tab, false),
     mnesia_lib:set({Tab, {index, Pos}}, Index),
-    add_index_info(Tab, SetOrBag, {Pos, {ram, Index}});
-add_ram_index(_Tab, snmp) ->
+    add_index_info(Tab, SetOrBag, {{Pos, Type}, {ram, Index}});
+add_ram_index(_Tab, _, snmp) ->
     ok.
 
-add_index_info(Tab, Type, IxElem) ->
+index_info(SetOrBag, PosList) ->
+    IxPlugins = mnesia_schema:index_plugins(),
+    ExpPosList = lists:map(
+		   fun({{P,Type},Ixt} = PI) ->
+			   case P of
+			       {_} = IxN ->
+				   {_, M, F} =
+				       lists:keyfind(IxN, 1, IxPlugins),
+				   {{{IxN,M,F}, Type}, Ixt};
+			       _ ->
+				   PI
+			   end
+		   end, PosList),
+    #index{setorbag = SetOrBag, pos_list = ExpPosList}.
+
+add_index_info(Tab, SetOrBag, IxElem) ->
     Commit = val({Tab, commit_work}),
     case lists:keysearch(index, 1, Commit) of
 	false ->
-	    Index = #index{setorbag = Type, 
-			   pos_list = [IxElem]},
-	    %% Check later if mnesia_tm is sensative about the order 
+	    IndexInfo = index_info(SetOrBag, [IxElem]),
+	    %% Check later if mnesia_tm is sensitive about the order
+	    mnesia_lib:set({Tab, index_info}, IndexInfo),
+	    mnesia_lib:set({Tab, index}, index_positions(IndexInfo)),
 	    mnesia_lib:set({Tab, commit_work}, 
-			   mnesia_lib:sort_commit([Index | Commit]));
+			   mnesia_lib:sort_commit([IndexInfo | Commit]));
 	{value, Old} ->
 	    %% We could check for consistency here
 	    Index = Old#index{pos_list = [IxElem | Old#index.pos_list]},
+	    mnesia_lib:set({Tab, index_info}, Index),
+	    mnesia_lib:set({Tab, index}, index_positions(Index)),
 	    NewC = lists:keyreplace(index, 1, Commit, Index),
 	    mnesia_lib:set({Tab, commit_work}, 
 			   mnesia_lib:sort_commit(NewC))
     end.
 
+index_positions(#index{pos_list = PL}) ->
+    [P || {{P,_},_} <- PL].
+
 del_index_info(Tab, Pos) ->
     Commit = val({Tab, commit_work}),
     case lists:keysearch(index, 1, Commit) of
@@ -345,13 +483,21 @@ del_index_info(Tab, Pos) ->
 	    %% Something is wrong ignore
 	    skip;
 	{value, Old} ->
-	    case lists:keydelete(Pos, 1, Old#index.pos_list) of
+	    case lists:filter(fun({P,_}) ->
+                                      element(1,P)=/=Pos
+                              end,
+                              Old#index.pos_list) of
 		[] -> 
+                    IndexInfo = index_info(Old#index.setorbag,[]),
+		    mnesia_lib:set({Tab, index_info}, IndexInfo),
+		    mnesia_lib:set({Tab, index}, index_positions(IndexInfo)),
 		    NewC = lists:keydelete(index, 1, Commit),
 		    mnesia_lib:set({Tab, commit_work}, 
 				   mnesia_lib:sort_commit(NewC));
 		New ->
 		    Index = Old#index{pos_list = New},
+		    mnesia_lib:set({Tab, index_info}, Index),
+		    mnesia_lib:set({Tab, index}, index_positions(Index)),
 		    NewC = lists:keyreplace(index, 1, Commit, Index),
 		    mnesia_lib:set({Tab, commit_work}, 
 				   mnesia_lib:sort_commit(NewC))
@@ -360,33 +506,69 @@ del_index_info(Tab, Pos) ->
 
 db_put({ram, Ixt}, V) ->
     true = ?ets_insert(Ixt, V);
+db_put({{ext, _, _} = Ext, Ixt}, V) ->
+    mnesia_lib:db_put(Ext, Ixt, V);
 db_put({dets, Ixt}, V) ->
     ok = dets:insert(Ixt, V).
 
-db_get({ram, Ixt}, K) ->
-    ?ets_lookup(Ixt, K);
+db_get({ram, _}=Ixt, IxKey) ->
+    Pat = [{{{IxKey, '$1'}}, [], [{{IxKey,'$1'}}]}],
+    db_select(Ixt, Pat);
+db_get({{ext,_,_} = _Storage, {_,_,{_,Type}}} = Ixt, IxKey) ->
+    Pat = case Type of
+	      ordered -> [{{{IxKey, '$1'}}, [], [{{IxKey,'$1'}}]}];
+	      bag     -> [{{IxKey, '_'}, [], ['$_']}]
+	  end,
+    db_select(Ixt, Pat);
 db_get({dets, Ixt}, K) ->
     dets:lookup(Ixt, K).
 
+db_erase({ram, Ixt}, K) ->
+    ?ets_delete(Ixt, K);
+db_erase({{ext,_,_} = Ext, Ixt}, K) ->
+    mnesia_lib:db_erase(Ext, Ixt, K);
+db_erase({dets, Ixt}, K) ->
+    dets:delete(Ixt, K).
+
 db_match_erase({ram, Ixt}, Pat) ->
     true = ?ets_match_delete(Ixt, Pat);
+db_match_erase({{ext,_,_} = Ext, Ixt}, Pat) ->
+    mnesia_lib:db_match_erase(Ext, Ixt, Pat);
 db_match_erase({dets, Ixt}, Pat) ->
     ok = dets:match_delete(Ixt, Pat).
     
-db_match({ram, Ixt}, Pat) ->
-    ?ets_match(Ixt, Pat);
-db_match({dets, Ixt}, Pat) ->
-    dets:match(Ixt, Pat).
+db_select({ram, Ixt}, Pat) ->
+    ets:select(Ixt, Pat);
+db_select({{ext,_,_} = Ext, Ixt}, Pat) ->
+    mnesia_lib:db_select(Ext, Ixt, Pat);
+db_select({dets, Ixt}, Pat) ->
+    dets:select(Ixt, Pat).
+
     
 get_index_table(Tab, Pos) ->
     get_index_table(Tab,  val({Tab, storage_type}), Pos).
 
-get_index_table(Tab, ram_copies, Pos) ->
-    {ram,  val({Tab, {index, Pos}})};
-get_index_table(Tab, disc_copies, Pos) ->
-    {ram,  val({Tab, {index, Pos}})};
-get_index_table(Tab, disc_only_copies, Pos) ->
-    {dets, val({Tab, {index, Pos}})};
-get_index_table(_Tab, unknown, _Pos) ->
-    unknown.
-
+get_index_table(Tab, _Storage, Pos) ->
+    #index{pos_list = PosL} = val({Tab, index_info}),
+    {_IxType, Ixt} = pick_index(PosL, Tab, Pos),
+    Ixt.
+
+index_vals_f(Storage, Tab, {_} = Pos) ->
+    index_vals_f(Storage, Tab,
+		 lists:keyfind(Pos, 1, mnesia_schema:index_plugins()));
+index_vals_f(_Storage, Tab, {Pos,M,F}) ->
+    fun(Obj) ->
+	    M:F(Tab, Pos, Obj)
+    end;
+index_vals_f(Storage, Tab, Pos) when is_integer(Pos) ->
+    case mnesia_lib:semantics(Storage, index_fun) of
+	undefined ->
+	    fun(Obj) ->
+		    [element(Pos, Obj)]
+	    end;
+	F when is_function(F, 4) ->
+	    {ext, Alias, _Mod} = Storage,
+	    fun(Obj) ->
+		    F(Alias, Tab, Pos, Obj)
+	    end
+    end.