yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

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

This commit is contained in:
JianBo He 2023-07-27 13:39:37 +08:00 committed by GitHub
parent 41e7c45a4b
commit 5e4855334e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 71 additions and 297 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
apps/emqx_rule_engine/src/emqx_rule_engine.erl
View File

@ -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()].

131
apps/emqx_rule_engine/src/emqx_rule_index.erl
View File

@ -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
%% parsed topic filters and record IDs, wrapped in a tuple to order them
%% strictly greater than unit tuple (`{}`). Existing table may be used if
%% existing keys will not collide with index keys.
%% Works on top of ETS ordered_set table. Keys are parsed topic filters
%% with record ID appended to the end, wrapped in a tuple to disambiguate from
%% topic filter words. Existing table may be used if existing keys will not
%% 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, {}}),
case match_filter(Prefix, K, Words == []) of
K = ets:next(Tab, Prefix),
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)].
matches(Topic, Tab, Opts) ->
-spec matches(emqx_types:topic(), ets:table()) -> [match(_ID)].
matches(Topic, Tab) ->
{Words, RPrefix} = match_init(Topic),
AccIn =
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, Tab).
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) ->
case match_filter(Prefix, K, Words == []) of
matches(K, Prefix, Words, RPrefix, Tab) ->
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, [W | RPrefix], Acc, Tab);
matches_rest(sharp, [W | Rest], RPrefix, Acc, Tab) ->
NAcc1 = matches([], ['#' | RPrefix], Acc, Tab),
NAcc2 = matches(Rest, ['+' | RPrefix], NAcc1, 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.
matches_rest(false, [W | Rest], RPrefix, Tab) ->
matches(Rest, [W | RPrefix], Tab);
matches_rest(plus, [W | Rest], RPrefix, Tab) ->
matches(Rest, ['+' | RPrefix], Tab) ++ matches(Rest, [W | RPrefix], Tab);
matches_rest(_, [], _RPrefix, _Tab) ->
[].
match_add(K = {_Filter, ID}, Acc = #{}) ->
Acc#{ID => K};
match_add(K, Acc) ->
[K | Acc].
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([], ['#']) ->
match_filter([], [{_ID}], _IsPrefix = false) ->
% NOTE: exact match is `true` only if we match whole topic, not prefix
true;
match_filter([], ['#', {_ID}], _IsPrefix) ->
% 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(T1, T2);
match_filter([H1 | _], [H2 | _]) when H2 > H1 ->
match_filter([H | T1], [H | T2], IsPrefix) ->
match_filter(T1, T2, IsPrefix);
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).

39
apps/emqx_rule_engine/test/emqx_rule_engine_SUITE.erl
View File

@ -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)),

196
apps/emqx_rule_engine/test/emqx_rule_index_SUITE.erl
View File

@ -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(t_insert_3, id(match(<<"sensor">>, Tab))),
?assertEqual(<<"sensor/#">>, get_topic(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) ->
emqx_rule_index:get_id(Match).
-spec get_id(emqx_rule_index:match(ID)) -> ID.
get_id([{ID}]) ->
ID;
get_id([_ | Rest]) ->
get_id(Rest).
topic(Match) ->
emqx_rule_index:get_topic(Match).
-spec get_topic(emqx_rule_index:match(_ID)) -> emqx_types:topic().
get_topic(K) ->
emqx_topic:join(cut_topic(K)).
topic_filter_to_topic_name(Topic) when is_binary(Topic) ->
topic_filter_to_topic_name(emqx_topic:words(Topic));
topic_filter_to_topic_name(Words) when is_list(Words) ->
topic_filter_to_topic_name(Words, []).
cut_topic(K) ->
cut_topic(K, []).
topic_filter_to_topic_name([], Acc) ->
emqx_topic:join(lists:reverse(Acc));
topic_filter_to_topic_name(['#' | _Rest], Acc) ->
case rand:uniform(2) of
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
)
).
cut_topic([{_ID}], Acc) ->
lists:reverse(Acc);
cut_topic([W | Rest], Acc) ->
cut_topic(Rest, [W | Acc]).