Merge pull request #11356 from emqx/revert-11332-fix-topic-index-unique-bugs

Revert "Fix(topicidx): allow to return matches unique by record id"

apps/emqx_rule_engine/src/emqx_rule_engine.erl

@@ -226,7 +226,7 @@ get_rules_ordered_by_ts() ->
 get_rules_for_topic(Topic) ->
     [
         emqx_rule_index:get_record(M, ?RULE_TOPIC_INDEX)
-     || M <- emqx_rule_index:matches(Topic, ?RULE_TOPIC_INDEX, [unique])
+     || M <- emqx_rule_index:matches(Topic, ?RULE_TOPIC_INDEX)
     ].
 
 -spec get_rules_with_same_event(Topic :: binary()) -> [rule()].

apps/emqx_rule_engine/src/emqx_rule_index.erl

@@ -16,29 +16,29 @@
 
 %% @doc Topic index for matching topics to topic filters.
 %%
-%% Works on top of ETS ordered_set table. Keys are tuples constructed from
+%% Works on top of ETS ordered_set table. Keys are parsed topic filters
-%% parsed topic filters and record IDs, wrapped in a tuple to order them
+%% with record ID appended to the end, wrapped in a tuple to disambiguate from
-%% strictly greater than unit tuple (`{}`). Existing table may be used if
+%% topic filter words. Existing table may be used if existing keys will not
-%% existing keys will not collide with index keys.
+%% collide with index keys.
 %%
 %% Designed to effectively answer questions like:
 %% 1. Does any topic filter match given topic?
 %% 2. Which records are associated with topic filters matching given topic?
-%% 3. Which topic filters match given topic?
+%%
-%% 4. Which record IDs are associated with topic filters matching given topic?
+%% Questions like these are _only slightly_ less effective:
+%% 1. Which topic filters match given topic?
+%% 2. Which record IDs are associated with topic filters matching given topic?
 
 -module(emqx_rule_index).
 
 -export([insert/4]).
 -export([delete/3]).
 -export([match/2]).
--export([matches/3]).
+-export([matches/2]).
 
--export([get_id/1]).
--export([get_topic/1]).
 -export([get_record/2]).
 
--type key(ID) :: {[binary() | '+' | '#'], {ID}}.
+-type key(ID) :: [binary() | '+' | '#' | {ID}].
 -type match(ID) :: key(ID).
 
 -ifdef(TEST).
@@ -46,10 +46,11 @@
 -endif.
 
 insert(Filter, ID, Record, Tab) ->
-    ets:insert(Tab, {{emqx_topic:words(Filter), {ID}}, Record}).
+    %% TODO: topic compact. see also in emqx_trie.erl
+    ets:insert(Tab, {emqx_topic:words(Filter) ++ [{ID}], Record}).
 
 delete(Filter, ID, Tab) ->
-    ets:delete(Tab, {emqx_topic:words(Filter), {ID}}).
+    ets:delete(Tab, emqx_topic:words(Filter) ++ [{ID}]).
 
 -spec match(emqx_types:topic(), ets:table()) -> match(_ID) | false.
 match(Topic, Tab) ->
@@ -58,8 +59,8 @@ match(Topic, Tab) ->
 
 match(Words, RPrefix, Tab) ->
     Prefix = lists:reverse(RPrefix),
-    K = ets:next(Tab, {Prefix, {}}),
+    K = ets:next(Tab, Prefix),
-    case match_filter(Prefix, K, Words == []) of
+    case match_filter(Prefix, K, Words =/= []) of
         true ->
             K;
         stop ->
@@ -72,7 +73,7 @@ match_rest(false, [W | Rest], RPrefix, Tab) ->
     match(Rest, [W | RPrefix], Tab);
 match_rest(plus, [W | Rest], RPrefix, Tab) ->
     case match(Rest, ['+' | RPrefix], Tab) of
-        Match = {_, _} ->
+        Match when is_list(Match) ->
             Match;
         false ->
             match(Rest, [W | RPrefix], Tab)
@@ -80,86 +81,48 @@ match_rest(plus, [W | Rest], RPrefix, Tab) ->
 match_rest(_, [], _RPrefix, _Tab) ->
     false.
 
--spec matches(emqx_types:topic(), ets:table(), _Opts :: [unique]) -> [match(_ID)].
+-spec matches(emqx_types:topic(), ets:table()) -> [match(_ID)].
-matches(Topic, Tab, Opts) ->
+matches(Topic, Tab) ->
     {Words, RPrefix} = match_init(Topic),
-    AccIn =
+    matches(Words, RPrefix, Tab).
-        case Opts of
-            [unique | _] -> #{};
-            [] -> []
-        end,
-    Matches = matches(Words, RPrefix, AccIn, Tab),
-    %% return rules ordered by Rule ID
-    case Matches of
-        #{} ->
-            maps:values(Matches);
-        _ ->
-            F = fun({_, {ID1}}, {_, {ID2}}) -> ID1 < ID2 end,
-            lists:sort(F, Matches)
-    end.
 
-matches(Words, RPrefix, Acc, Tab) ->
+matches(Words, RPrefix, Tab) ->
     Prefix = lists:reverse(RPrefix),
-    matches(ets:next(Tab, {Prefix, {}}), Prefix, Words, RPrefix, Acc, Tab).
+    matches(ets:next(Tab, Prefix), Prefix, Words, RPrefix, Tab).
 
-matches(K, Prefix, Words, RPrefix, Acc, Tab) ->
+matches(K, Prefix, Words, RPrefix, Tab) ->
-    case match_filter(Prefix, K, Words == []) of
+    case match_filter(Prefix, K, Words =/= []) of
         true ->
-            matches(ets:next(Tab, K), Prefix, Words, RPrefix, match_add(K, Acc), Tab);
+            [K | matches(ets:next(Tab, K), Prefix, Words, RPrefix, Tab)];
         stop ->
-            Acc;
+            [];
         Matched ->
-            matches_rest(Matched, Words, RPrefix, Acc, Tab)
+            matches_rest(Matched, Words, RPrefix, Tab)
     end.
 
-matches_rest(false, [W | Rest], RPrefix, Acc, Tab) ->
+matches_rest(false, [W | Rest], RPrefix, Tab) ->
-    matches(Rest, [W | RPrefix], Acc, Tab);
+    matches(Rest, [W | RPrefix], Tab);
-matches_rest(sharp, [W | Rest], RPrefix, Acc, Tab) ->
+matches_rest(plus, [W | Rest], RPrefix, Tab) ->
-    NAcc1 = matches([], ['#' | RPrefix], Acc, Tab),
+    matches(Rest, ['+' | RPrefix], Tab) ++ matches(Rest, [W | RPrefix], Tab);
-    NAcc2 = matches(Rest, ['+' | RPrefix], NAcc1, Tab),
+matches_rest(_, [], _RPrefix, _Tab) ->
-    matches(Rest, [W | RPrefix], NAcc2, Tab);
+    [].
-matches_rest(plus, [W | Rest], RPrefix, Acc, Tab) ->
-    NAcc = matches(Rest, ['+' | RPrefix], Acc, Tab),
-    matches(Rest, [W | RPrefix], NAcc, Tab);
-matches_rest(_, [], _RPrefix, Acc, _Tab) ->
-    Acc.
 
-match_add(K = {_Filter, ID}, Acc = #{}) ->
+match_filter([], [{_ID}], _IsPrefix = false) ->
-    Acc#{ID => K};
+    % NOTE: exact match is `true` only if we match whole topic, not prefix
-match_add(K, Acc) ->
+    true;
-    [K | Acc].
+match_filter([], ['#', {_ID}], _IsPrefix) ->
-
-match_filter(Prefix, {Filter, _ID}, NotPrefix) ->
-    case match_filter(Prefix, Filter) of
-        exact ->
-            % NOTE: exact match is `true` only if we match whole topic, not prefix
-            case NotPrefix of
-                true ->
-                    true;
-                false ->
-                    sharp
-            end;
-        Match ->
-            Match
-    end;
-match_filter(_, '$end_of_table', _) ->
-    stop.
-
-match_filter([], []) ->
-    exact;
-match_filter([], ['' | _]) ->
-    sharp;
-match_filter([], ['#']) ->
     % NOTE: naturally, '#' < '+', so this is already optimal for `match/2`
     true;
-match_filter([], ['+' | _]) ->
+match_filter([], ['+' | _], _) ->
     plus;
-match_filter([], [_H | _]) ->
+match_filter([], [_H | _], _) ->
     false;
-match_filter([H | T1], [H | T2]) ->
+match_filter([H | T1], [H | T2], IsPrefix) ->
-    match_filter(T1, T2);
+    match_filter(T1, T2, IsPrefix);
-match_filter([H1 | _], [H2 | _]) when H2 > H1 ->
+match_filter([H1 | _], [H2 | _], _) when H2 > H1 ->
     % NOTE: we're strictly past the prefix, no need to continue
+    stop;
+match_filter(_, '$end_of_table', _) ->
     stop.
 
 match_init(Topic) ->
@@ -172,14 +135,6 @@ match_init(Topic) ->
             {Words, []}
     end.
 
--spec get_id(match(ID)) -> ID.
-get_id({_Filter, {ID}}) ->
-    ID.
-
--spec get_topic(match(_ID)) -> emqx_types:topic().
-get_topic({Filter, _ID}) ->
-    emqx_topic:join(Filter).
-
 -spec get_record(match(_ID), ets:table()) -> _Record.
 get_record(K, Tab) ->
     ets:lookup_element(Tab, K, 2).

apps/emqx_rule_engine/test/emqx_rule_engine_SUITE.erl

@@ -58,7 +58,6 @@ groups() ->
             t_create_existing_rule,
             t_get_rules_for_topic,
             t_get_rules_for_topic_2,
-            t_get_rules_for_topic_3,
             t_get_rules_with_same_event,
             t_get_rule_ids_by_action,
             t_ensure_action_removed
@@ -401,44 +400,6 @@ t_get_rules_for_topic_2(_Config) ->
     ]),
     ok.
 
-t_get_rules_for_topic_3(_Config) ->
-    ok = create_rules(
-        [
-            make_simple_rule(<<"rule-debug-5">>, <<"select * from \"simple/#\"">>),
-            make_simple_rule(<<"rule-debug-4">>, <<"select * from \"simple/+\"">>),
-            make_simple_rule(<<"rule-debug-3">>, <<"select * from \"simple/+/1\"">>),
-            make_simple_rule(<<"rule-debug-2">>, <<"select * from \"simple/1\"">>),
-            make_simple_rule(
-                <<"rule-debug-1">>,
-                <<"select * from \"simple/2\", \"simple/+\", \"simple/3\"">>
-            )
-        ]
-    ),
-    Rules1 = get_rules_for_topic_in_e510_impl(<<"simple/1">>),
-    Rules2 = emqx_rule_engine:get_rules_for_topic(<<"simple/1">>),
-    %% assert, ensure the order of rules is the same as e5.1.0
-    ?assertEqual(Rules1, Rules2),
-    ?assertEqual(
-        [<<"rule-debug-1">>, <<"rule-debug-2">>, <<"rule-debug-4">>, <<"rule-debug-5">>],
-        [Id || #{id := Id} <- Rules1]
-    ),
-
-    ok = delete_rules_by_ids([
-        <<"rule-debug-1">>,
-        <<"rule-debug-2">>,
-        <<"rule-debug-3">>,
-        <<"rule-debug-4">>,
-        <<"rule-debug-5">>
-    ]),
-    ok.
-
-get_rules_for_topic_in_e510_impl(Topic) ->
-    [
-        Rule
-     || Rule = #{from := From} <- emqx_rule_engine:get_rules(),
-        emqx_topic:match_any(Topic, From)
-    ].
-
 t_get_rules_with_same_event(_Config) ->
     PubT = <<"simple/1">>,
     PubN = length(emqx_rule_engine:get_rules_with_same_event(PubT)),

apps/emqx_rule_engine/test/emqx_rule_index_SUITE.erl

@@ -19,7 +19,6 @@
 -compile(export_all).
 -compile(nowarn_export_all).
 
--include_lib("proper/include/proper.hrl").
 -include_lib("eunit/include/eunit.hrl").
 
 all() ->
@@ -30,8 +29,8 @@ t_insert(_) ->
     true = emqx_rule_index:insert(<<"sensor/1/metric/2">>, t_insert_1, <<>>, Tab),
     true = emqx_rule_index:insert(<<"sensor/+/#">>, t_insert_2, <<>>, Tab),
     true = emqx_rule_index:insert(<<"sensor/#">>, t_insert_3, <<>>, Tab),
-    ?assertEqual(<<"sensor/#">>, topic(match(<<"sensor">>, Tab))),
+    ?assertEqual(<<"sensor/#">>, get_topic(match(<<"sensor">>, Tab))),
-    ?assertEqual(t_insert_3, id(match(<<"sensor">>, Tab))),
+    ?assertEqual(t_insert_3, get_id(match(<<"sensor">>, Tab))),
     true = ets:delete(Tab).
 
 t_match(_) ->
@@ -41,7 +40,7 @@ t_match(_) ->
     true = emqx_rule_index:insert(<<"sensor/#">>, t_match_3, <<>>, Tab),
     ?assertMatch(
         [<<"sensor/#">>, <<"sensor/+/#">>],
-        [topic(M) || M <- matches(<<"sensor/1">>, Tab)]
+        [get_topic(M) || M <- matches(<<"sensor/1">>, Tab)]
     ),
     true = ets:delete(Tab).
 
@@ -52,7 +51,7 @@ t_match2(_) ->
     true = emqx_rule_index:insert(<<"+/+/#">>, t_match2_3, <<>>, Tab),
     ?assertEqual(
         [<<"#">>, <<"+/#">>, <<"+/+/#">>],
-        [topic(M) || M <- matches(<<"a/b/c">>, Tab)]
+        [get_topic(M) || M <- matches(<<"a/b/c">>, Tab)]
     ),
     ?assertEqual(
         false,
@@ -80,7 +79,7 @@ t_match3(_) ->
     end,
     ?assertEqual(
         t_match3_sys,
-        id(match(<<"$SYS/a/b/c">>, Tab))
+        get_id(match(<<"$SYS/a/b/c">>, Tab))
     ),
     true = ets:delete(Tab).
 
@@ -93,11 +92,11 @@ t_match4(_) ->
     ),
     ?assertEqual(
         [<<"/#">>, <<"/+">>],
-        [topic(M) || M <- matches(<<"/">>, Tab)]
+        [get_topic(M) || M <- matches(<<"/">>, Tab)]
     ),
     ?assertEqual(
         [<<"/#">>, <<"/+/a/b/c">>],
-        [topic(M) || M <- matches(<<"/0/a/b/c">>, Tab)]
+        [get_topic(M) || M <- matches(<<"/0/a/b/c">>, Tab)]
     ),
     true = ets:delete(Tab).
 
@@ -115,11 +114,11 @@ t_match5(_) ->
     ),
     ?assertEqual(
         [<<"#">>, <<T/binary, "/#">>],
-        [topic(M) || M <- matches(T, Tab)]
+        [get_topic(M) || M <- matches(T, Tab)]
     ),
     ?assertEqual(
         [<<"#">>, <<T/binary, "/#">>, <<T/binary, "/+">>],
-        [topic(M) || M <- matches(<<T/binary, "/1">>, Tab)]
+        [get_topic(M) || M <- matches(<<T/binary, "/1">>, Tab)]
     ),
     true = ets:delete(Tab).
 
@@ -128,7 +127,7 @@ t_match6(_) ->
     T = <<"a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/z">>,
     W = <<"+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/#">>,
     emqx_rule_index:insert(W, ID = t_match6, <<>>, Tab),
-    ?assertEqual(ID, id(match(T, Tab))),
+    ?assertEqual(ID, get_id(match(T, Tab))),
     true = ets:delete(Tab).
 
 t_match7(_) ->
@@ -136,173 +135,32 @@ t_match7(_) ->
     T = <<"a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/z">>,
     W = <<"a/+/c/+/e/+/g/+/i/+/k/+/m/+/o/+/q/+/s/+/u/+/w/+/y/+/#">>,
     emqx_rule_index:insert(W, t_match7, <<>>, Tab),
-    ?assertEqual(W, topic(match(T, Tab))),
+    ?assertEqual(W, get_topic(match(T, Tab))),
     true = ets:delete(Tab).
 
-t_match_unique(_) ->
-    Tab = new(),
-    emqx_rule_index:insert(<<"a/b/c">>, t_match_id1, <<>>, Tab),
-    emqx_rule_index:insert(<<"a/b/+">>, t_match_id1, <<>>, Tab),
-    emqx_rule_index:insert(<<"a/b/c/+">>, t_match_id2, <<>>, Tab),
-    ?assertEqual(
-        [t_match_id1, t_match_id1],
-        [id(M) || M <- emqx_rule_index:matches(<<"a/b/c">>, Tab, [])]
-    ),
-    ?assertEqual(
-        [t_match_id1],
-        [id(M) || M <- emqx_rule_index:matches(<<"a/b/c">>, Tab, [unique])]
-    ).
-
-t_match_ordering(_) ->
-    Tab = new(),
-    emqx_rule_index:insert(<<"a/b/+">>, t_match_id2, <<>>, Tab),
-    emqx_rule_index:insert(<<"a/b/c">>, t_match_id1, <<>>, Tab),
-    emqx_rule_index:insert(<<"a/b/#">>, t_match_id3, <<>>, Tab),
-    Ids1 = [id(M) || M <- emqx_rule_index:matches(<<"a/b/c">>, Tab, [])],
-    Ids2 = [id(M) || M <- emqx_rule_index:matches(<<"a/b/c">>, Tab, [unique])],
-    ?assertEqual(Ids1, Ids2),
-    ?assertEqual([t_match_id1, t_match_id2, t_match_id3], Ids1).
-
-t_match_wildcard_edge_cases(_) ->
-    CommonTopics = [
-        <<"a/b">>,
-        <<"a/b/#">>,
-        <<"a/b/#">>,
-        <<"a/b/c">>,
-        <<"a/b/+">>,
-        <<"a/b/d">>,
-        <<"a/+/+">>,
-        <<"a/+/#">>
-    ],
-    Datasets =
-        [
-            %% Topics, TopicName, Results
-            {CommonTopics, <<"a/b/c">>, [2, 3, 4, 5, 7, 8]},
-            {CommonTopics, <<"a/b">>, [1, 2, 3, 8]},
-            {[<<"+/b/c">>, <<"/">>], <<"a/b/c">>, [1]},
-            {[<<"#">>, <<"/">>], <<"a">>, [1]},
-            {[<<"/">>, <<"+">>], <<"a">>, [2]}
-        ],
-    F = fun({Topics, TopicName, Expected}) ->
-        Tab = new(),
-        _ = lists:foldl(
-            fun(T, N) ->
-                emqx_rule_index:insert(T, N, <<>>, Tab),
-                N + 1
-            end,
-            1,
-            Topics
-        ),
-        Results = [id(M) || M <- emqx_rule_index:matches(TopicName, Tab, [unique])],
-        case Results == Expected of
-            true ->
-                ets:delete(Tab);
-            false ->
-                ct:pal(
-                    "Base topics: ~p~n"
-                    "Topic name: ~p~n"
-                    "Index results: ~p~n"
-                    "Expected results:: ~p~n",
-                    [Topics, TopicName, Results, Expected]
-                ),
-                error(bad_matches)
-        end
-    end,
-    lists:foreach(F, Datasets).
-
-t_prop_matches(_) ->
-    ?assert(
-        proper:quickcheck(
-            topic_matches_prop(),
-            [{max_size, 100}, {numtests, 100}]
-        )
-    ).
-
-topic_matches_prop() ->
-    ?FORALL(
-        Topics0,
-        list(emqx_proper_types:topic()),
-        begin
-            Tab = new(),
-            Topics = lists:filter(fun(Topic) -> Topic =/= <<>> end, Topics0),
-            lists:foldl(
-                fun(Topic, N) ->
-                    true = emqx_rule_index:insert(Topic, N, <<>>, Tab),
-                    N + 1
-                end,
-                1,
-                Topics
-            ),
-            lists:foreach(
-                fun(Topic0) ->
-                    Topic = topic_filter_to_topic_name(Topic0),
-                    Ids1 = [
-                        emqx_rule_index:get_id(R)
-                     || R <- emqx_rule_index:matches(Topic, Tab, [unique])
-                    ],
-                    Ids2 = topic_matches(Topic, Topics),
-                    case Ids2 == Ids1 of
-                        true ->
-                            ok;
-                        false ->
-                            ct:pal(
-                                "Base topics: ~p~n"
-                                "Topic name: ~p~n"
-                                "Index results: ~p~n"
-                                "Topic match results:: ~p~n",
-                                [Topics, Topic, Ids1, Ids2]
-                            ),
-                            error(bad_matches)
-                    end
-                end,
-                Topics
-            ),
-            true
-        end
-    ).
-
-%%--------------------------------------------------------------------
-%% helpers
-
 new() ->
-    ets:new(?MODULE, [public, ordered_set, {read_concurrency, true}]).
+    ets:new(?MODULE, [public, ordered_set, {write_concurrency, true}]).
 
 match(T, Tab) ->
     emqx_rule_index:match(T, Tab).
 
 matches(T, Tab) ->
-    lists:sort(emqx_rule_index:matches(T, Tab, [])).
+    lists:sort(emqx_rule_index:matches(T, Tab)).
 
-id(Match) ->
+-spec get_id(emqx_rule_index:match(ID)) -> ID.
-    emqx_rule_index:get_id(Match).
+get_id([{ID}]) ->
+    ID;
+get_id([_ | Rest]) ->
+    get_id(Rest).
 
-topic(Match) ->
+-spec get_topic(emqx_rule_index:match(_ID)) -> emqx_types:topic().
-    emqx_rule_index:get_topic(Match).
+get_topic(K) ->
+    emqx_topic:join(cut_topic(K)).
 
-topic_filter_to_topic_name(Topic) when is_binary(Topic) ->
+cut_topic(K) ->
-    topic_filter_to_topic_name(emqx_topic:words(Topic));
+    cut_topic(K, []).
-topic_filter_to_topic_name(Words) when is_list(Words) ->
-    topic_filter_to_topic_name(Words, []).
 
-topic_filter_to_topic_name([], Acc) ->
+cut_topic([{_ID}], Acc) ->
-    emqx_topic:join(lists:reverse(Acc));
+    lists:reverse(Acc);
-topic_filter_to_topic_name(['#' | _Rest], Acc) ->
+cut_topic([W | Rest], Acc) ->
-    case rand:uniform(2) of
+    cut_topic(Rest, [W | Acc]).
-        1 -> emqx_topic:join(lists:reverse(Acc));
-        _ -> emqx_topic:join(lists:reverse([<<"_sharp">> | Acc]))
-    end;
-topic_filter_to_topic_name(['+' | Rest], Acc) ->
-    topic_filter_to_topic_name(Rest, [<<"_plus">> | Acc]);
-topic_filter_to_topic_name([H | Rest], Acc) ->
-    topic_filter_to_topic_name(Rest, [H | Acc]).
-
-topic_matches(Topic, Topics0) ->
-    Topics = lists:zip(lists:seq(1, length(Topics0)), Topics0),
-    lists:sort(
-        lists:filtermap(
-            fun({Id, Topic0}) ->
-                emqx_topic:match(Topic, Topic0) andalso {true, Id}
-            end,
-            Topics
-        )
-    ).