Fix erroneous aborts when using index plugins (ERL-556)

- first stab at fixing index_[read|match_object] w plugins
- do not include index plugins at match ops
- add index_plugin_test_suite
- Increase ix plugin test coverage + bug fix

2 files changed, 94 insertions, 24 deletions

diff --git a/lib/mnesia/src/mnesia.erl b/lib/mnesia/src/mnesia.erl
index 77afb8250c..02bc884e36 100644
--- a/lib/mnesia/src/mnesia.erl
+++ b/lib/mnesia/src/mnesia.erl
@@ -160,7 +160,7 @@
                     {'sync_transaction', Retries::non_neg_integer()}.
 -type table() :: atom().
 -type storage_type() :: 'ram_copies' | 'disc_copies' | 'disc_only_copies'.
--type index_attr() :: atom() | non_neg_integer().
+-type index_attr() :: atom() | non_neg_integer() | {atom()}.
 -type write_locks() :: 'write' | 'sticky_write'.
 -type read_locks() :: 'read'.
 -type lock_kind() :: write_locks() | read_locks().
@@ -1277,6 +1277,14 @@ match_object(Tid, Ts, Tab, Pat, LockKind)
 match_object(_Tid, _Ts, Tab, Pat, _LockKind) ->
     abort({bad_type, Tab, Pat}).
 
+add_written_index(Store, Pos, Tab, Key, Objs) when is_integer(Pos) ->
+    Pat = setelement(Pos, val({Tab, wild_pattern}), Key),
+    add_written_match(Store, Pat, Tab, Objs);
+add_written_index(Store, Pos, Tab, Key, Objs) when is_tuple(Pos) ->
+    IxF = mnesia_index:index_vals_f(val({Tab, storage_type}), Tab, Pos),
+    Ops = find_ops(Store, Tab, '_'),
+    add_ix_match(Ops, Objs, IxF, Key, val({Tab, setorbag})).
+
 add_written_match(S, Pat, Tab, Objs) ->
     Ops = find_ops(S, Tab, Pat),
     FixedRes = add_match(Ops, Objs, val({Tab, setorbag})),
@@ -1303,6 +1311,46 @@ add_match([{_Oid, Val, write}|R], Objs, bag) ->
 add_match([{Oid, Val, write}|R], Objs, set) ->
     add_match(R, [Val | deloid(Oid,Objs)],set).
 
+add_ix_match([], Objs, _IxF, _Key, _Type) ->
+    Objs;
+add_ix_match(Written, Objs, IxF, Key, ordered_set) ->
+    %% Must use keysort which is stable
+    add_ordered_match(lists:keysort(1, ix_filter_ops(IxF, Key, Written)), Objs, []);
+add_ix_match([{Oid, _, delete}|R], Objs, IxF, Key, Type) ->
+    add_ix_match(R, deloid(Oid, Objs), IxF, Key, Type);
+add_ix_match([{_Oid, Val, delete_object}|R], Objs, IxF, Key, Type) ->
+    case ix_match(Val, IxF, Key) of
+        true ->
+            add_ix_match(R, lists:delete(Val, Objs), IxF, Key, Type);
+        false ->
+            add_ix_match(R, Objs, IxF, Key, Type)
+    end;
+add_ix_match([{_Oid, Val, write}|R], Objs, IxF, Key, bag) ->
+    case ix_match(Val, IxF, Key) of
+        true ->
+            add_ix_match(R, [Val | lists:delete(Val, Objs)], IxF, Key, bag);
+        false ->
+            add_ix_match(R, Objs, IxF, Key, bag)
+    end;
+add_ix_match([{Oid, Val, write}|R], Objs, IxF, Key, set) ->
+    case ix_match(Val, IxF, Key) of
+        true ->
+            add_ix_match(R, [Val | deloid(Oid,Objs)],IxF,Key,set);
+        false ->
+            add_ix_match(R, Objs, IxF, Key, set)
+    end.
+
+ix_match(Val, IxF, Key) ->
+    lists:member(Key, IxF(Val)).
+
+ix_filter_ops(IxF, Key, Ops) ->
+    lists:filter(
+      fun({_Oid, Obj, write}) ->
+              ix_match(Obj, IxF, Key);
+         (_) ->
+              true
+      end, Ops).
+
 %% For ordered_set only !!
 add_ordered_match(Written = [{{_, Key}, _, _}|_], [Obj|Objs], Acc)
   when Key > element(2, Obj) ->
@@ -1641,6 +1689,16 @@ index_match_object(Tid, Ts, Tab, Pat, Attr, LockKind)
 	    dirty_index_match_object(Tab, Pat, Attr); % Should be optimized?
 	tid ->
 	    case mnesia_schema:attr_tab_to_pos(Tab, Attr) of
+                {_} ->
+                    case LockKind of
+                        read ->
+			    Store = Ts#tidstore.store,
+			    mnesia_locker:rlock_table(Tid, Store, Tab),
+			    Objs = dirty_match_object(Tab, Pat),
+			    add_written_match(Store, Pat, Tab, Objs);
+                        _ ->
+                            abort({bad_type, Tab, LockKind})
+                    end;
 		Pos when Pos =< tuple_size(Pat) ->
 		    case LockKind of
 			read ->
@@ -1688,8 +1746,8 @@ index_read(Tid, Ts, Tab, Key, Attr, LockKind)
 			false ->
 			    Store = Ts#tidstore.store,
 			    Objs = mnesia_index:read(Tid, Store, Tab, Key, Pos),
-			    Pat = setelement(Pos, val({Tab, wild_pattern}), Key),
-			    add_written_match(Store, Pat, Tab, Objs);
+                            add_written_index(
+                              Ts#tidstore.store, Pos, Tab, Key, Objs);
 			true ->
 			    abort({bad_type, Tab, Attr, Key})
 		    end;
@@ -1825,7 +1883,7 @@ remote_dirty_match_object(Tab, Pat) ->
 	false ->
 	    mnesia_lib:db_match_object(Tab, Pat);
 	true ->
-	    PosList = val({Tab, index}),
+            PosList = regular_indexes(Tab),
 	    remote_dirty_match_object(Tab, Pat, PosList)
     end.
 
@@ -1857,7 +1915,7 @@ remote_dirty_select(Tab, Spec) ->
 		false ->
 		    mnesia_lib:db_select(Tab, Spec);
 		true  ->
-		    PosList = val({Tab, index}),
+		    PosList = regular_indexes(Tab),
 		    remote_dirty_select(Tab, Spec, PosList)
 	    end;
 	_ ->
@@ -1924,6 +1982,8 @@ dirty_index_match_object(Pat, _Attr) ->
 dirty_index_match_object(Tab, Pat, Attr)
   when is_atom(Tab), Tab /= schema, is_tuple(Pat), tuple_size(Pat) > 2 ->
     case mnesia_schema:attr_tab_to_pos(Tab, Attr) of
+        {_} ->
+            dirty_match_object(Tab, Pat);
 	Pos when Pos =< tuple_size(Pat) ->
 	    case has_var(element(2, Pat)) of
 		false ->
@@ -3254,3 +3314,7 @@ put_activity_id(Activity) ->
     mnesia_tm:put_activity_id(Activity).
 put_activity_id(Activity,Fun) ->
     mnesia_tm:put_activity_id(Activity,Fun).
+
+regular_indexes(Tab) ->
+    PosList = val({Tab, index}),
+    [P || P <- PosList, is_integer(P)].
diff --git a/lib/mnesia/src/mnesia_index.erl b/lib/mnesia/src/mnesia_index.erl
index 098265d5fc..6f1c21e3b9 100644
--- a/lib/mnesia/src/mnesia_index.erl
+++ b/lib/mnesia/src/mnesia_index.erl
@@ -1,8 +1,8 @@
 %%
 %% %CopyrightBegin%
-%% 
+%%
 %% Copyright Ericsson AB 1996-2018. All Rights Reserved.
-%% 
+%%
 %% Licensed under the Apache License, Version 2.0 (the "License");
 %% you may not use this file except in compliance with the License.
 %% You may obtain a copy of the License at
@@ -14,7 +14,7 @@
 %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 %% See the License for the specific language governing permissions and
 %% limitations under the License.
-%% 
+%%
 %% %CopyrightEnd%
 %%
 
@@ -37,7 +37,7 @@
 	 db_match_erase/2,
 	 get_index_table/2,
 	 get_index_table/3,
-	 
+
 	 tab2filename/2,
 	 init_index/2,
 	 init_indecies/3,
@@ -45,6 +45,7 @@
 	 del_transient/3,
 	 del_index_table/3,
 
+         index_vals_f/3,
          index_info/2,
 	 ext_index_instances/1]).
 
@@ -60,9 +61,14 @@ read(Tid, Store, Tab, IxKey, Pos) ->
     ResList = mnesia_locker:ixrlock(Tid, Store, Tab, IxKey, Pos),
     %% Remove all tuples which don't include Ixkey, happens when Tab is a bag
     case val({Tab, setorbag}) of
-	bag -> 
+	bag when is_integer(Pos) ->
 	    mnesia_lib:key_search_all(IxKey, Pos, ResList);
-	_ -> 
+        bag when is_tuple(Pos) ->
+            TabStorage = val({Tab, storage_type}),
+            ValsF = index_vals_f(TabStorage, Tab, Pos),
+            [Obj || Obj <- ResList,
+                    lists:member(IxKey, ValsF(Obj))];
+	_ ->
 	    ResList
     end.
 
@@ -136,7 +142,7 @@ 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 -> 
+	bag ->
 	    del_object_bag(Type, ValsF, Tab, K, Obj, Ixt);
 	_ -> %% If set remove the tuple in index table
 	    del_ixes(Type, Ixt, ValsF, Obj, K)
@@ -197,7 +203,7 @@ merge([], _, _, Ack) ->
 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) when is_integer(Pos) ->
     %% Assume that we are on the node where the replica is
     %% Returns the records without applying the match spec
@@ -233,7 +239,7 @@ dirty_read2(Tab, IxKey, Pos) ->
 		  end, Acc, mnesia_lib:db_get(Storage, Tab, K))
 	end, [], Keys)).
 
-pick_index([{{{Pfx,_},IxType}, Ixt}|_], _Tab, {_} = Pfx) ->
+pick_index([{{{Pfx,_,_},IxType}, Ixt}|_], _Tab, {_} = Pfx) ->
     {IxType, Ixt};
 pick_index([{{Pos,IxType}, Ixt}|_], _Tab, Pos) ->
     {IxType, Ixt};
@@ -242,7 +248,7 @@ 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
@@ -387,12 +393,12 @@ init_ext_index(Tab, Storage, Alias, Mod, [{Pos,Type} | Tail]) ->
 create_fun(Cont, Tab, Pos) ->
     IxF = index_vals_f(disc_only_copies, Tab, Pos),
     fun(read) ->
-	    Data = 
+	    Data =
 		case Cont of
 		    {start, KeysPerChunk} ->
 			mnesia_lib:db_init_chunk(
 			  disc_only_copies, Tab, KeysPerChunk);
-		    '$end_of_table' -> 
+		    '$end_of_table' ->
 			'$end_of_table';
 		    _Else ->
 			mnesia_lib:db_chunk(disc_only_copies, Cont)
@@ -462,7 +468,7 @@ add_index_info(Tab, 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:set({Tab, commit_work},
 			   mnesia_lib:sort_commit([IndexInfo | Commit]));
 	{value, Old} ->
 	    %% We could check for consistency here
@@ -470,7 +476,7 @@ add_index_info(Tab, SetOrBag, IxElem) ->
 	    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:set({Tab, commit_work},
 			   mnesia_lib:sort_commit(NewC))
     end.
 
@@ -488,19 +494,19 @@ del_index_info(Tab, Pos) ->
                                       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: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:set({Tab, commit_work},
 				   mnesia_lib:sort_commit(NewC))
 	    end
     end.
@@ -537,7 +543,7 @@ 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_select({ram, Ixt}, Pat) ->
     ets:select(Ixt, Pat);
 db_select({{ext,_,_} = Ext, Ixt}, Pat) ->
@@ -545,7 +551,7 @@ db_select({{ext,_,_} = 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).