Merge pull request #11282 from JimMoen/perf-get-all-rule

feat(index): add topic index facility

apps/emqx_rule_engine/include/rule_engine.hrl

@@ -109,6 +109,7 @@
 
 %% Tables
 -define(RULE_TAB, emqx_rule_engine).
+-define(RULE_TOPIC_INDEX, emqx_rule_engine_topic_index).
 
 %% Allowed sql function provider modules
 -define(DEFAULT_SQL_FUNC_PROVIDER, emqx_rule_funcs).

apps/emqx_rule_engine/src/emqx_rule_engine.erl

@@ -176,7 +176,7 @@ create_rule(Params) ->
 
 create_rule(Params = #{id := RuleId}, CreatedAt) when is_binary(RuleId) ->
     case get_rule(RuleId) of
-        not_found -> parse_and_insert(Params, CreatedAt);
+        not_found -> with_parsed_rule(Params, CreatedAt, fun insert_rule/1);
         {ok, _} -> {error, already_exists}
     end.
 
@@ -185,18 +185,27 @@ update_rule(Params = #{id := RuleId}) when is_binary(RuleId) ->
     case get_rule(RuleId) of
         not_found ->
             {error, not_found};
-        {ok, #{created_at := CreatedAt}} ->
-            parse_and_insert(Params, CreatedAt)
+        {ok, RulePrev = #{created_at := CreatedAt}} ->
+            with_parsed_rule(Params, CreatedAt, fun(Rule) -> update_rule(Rule, RulePrev) end)
     end.
 
 -spec delete_rule(RuleId :: rule_id()) -> ok.
 delete_rule(RuleId) when is_binary(RuleId) ->
-    gen_server:call(?RULE_ENGINE, {delete_rule, RuleId}, ?T_CALL).
+    case get_rule(RuleId) of
+        not_found ->
+            ok;
+        {ok, Rule} ->
+            gen_server:call(?RULE_ENGINE, {delete_rule, Rule}, ?T_CALL)
+    end.
 
 -spec insert_rule(Rule :: rule()) -> ok.
 insert_rule(Rule) ->
     gen_server:call(?RULE_ENGINE, {insert_rule, Rule}, ?T_CALL).
 
+-spec update_rule(Rule :: rule(), RulePrev :: rule()) -> ok.
+update_rule(Rule, RulePrev) ->
+    gen_server:call(?RULE_ENGINE, {update_rule, Rule, RulePrev}, ?T_CALL).
+
 %%----------------------------------------------------------------------------------------
 %% Rule Management
 %%----------------------------------------------------------------------------------------
@@ -216,13 +225,13 @@ get_rules_ordered_by_ts() ->
 -spec get_rules_for_topic(Topic :: binary()) -> [rule()].
 get_rules_for_topic(Topic) ->
     [
-        Rule
-     || Rule = #{from := From} <- get_rules(),
-        emqx_topic:match_any(Topic, From)
+        emqx_rule_index:get_record(M, ?RULE_TOPIC_INDEX)
+     || M <- emqx_rule_index:matches(Topic, ?RULE_TOPIC_INDEX)
     ].
 
 -spec get_rules_with_same_event(Topic :: binary()) -> [rule()].
 get_rules_with_same_event(Topic) ->
+    %% TODO: event matching index not implemented yet
     EventName = emqx_rule_events:event_name(Topic),
     [
         Rule
@@ -232,6 +241,7 @@ get_rules_with_same_event(Topic) ->
 
 -spec get_rule_ids_by_action(action_name()) -> [rule_id()].
 get_rule_ids_by_action(BridgeId) when is_binary(BridgeId) ->
+    %% TODO: bridge matching index not implemented yet
     [
         Id
      || #{actions := Acts, id := Id, from := Froms} <- get_rules(),
@@ -239,6 +249,7 @@ get_rule_ids_by_action(BridgeId) when is_binary(BridgeId) ->
             references_ingress_bridge(Froms, BridgeId)
     ];
 get_rule_ids_by_action(#{function := FuncName}) when is_binary(FuncName) ->
+    %% TODO: action id matching index not implemented yet
     {Mod, Fun} =
         case string:split(FuncName, ":", leading) of
             [M, F] -> {binary_to_module(M), F};
@@ -411,10 +422,17 @@ init([]) ->
     {ok, #{}}.
 
 handle_call({insert_rule, Rule}, _From, State) ->
-    do_insert_rule(Rule),
+    ok = do_insert_rule(Rule),
+    ok = do_update_rule_index(Rule),
+    {reply, ok, State};
+handle_call({update_rule, Rule, RulePrev}, _From, State) ->
+    ok = do_delete_rule_index(RulePrev),
+    ok = do_insert_rule(Rule),
+    ok = do_update_rule_index(Rule),
     {reply, ok, State};
 handle_call({delete_rule, Rule}, _From, State) ->
-    do_delete_rule(Rule),
+    ok = do_delete_rule_index(Rule),
+    ok = do_delete_rule(Rule),
     {reply, ok, State};
 handle_call(Req, _From, State) ->
     ?SLOG(error, #{msg => "unexpected_call", request => Req}),
@@ -438,7 +456,7 @@ code_change(_OldVsn, State, _Extra) ->
 %% Internal Functions
 %%----------------------------------------------------------------------------------------
 
-parse_and_insert(Params = #{id := RuleId, sql := Sql, actions := Actions}, CreatedAt) ->
+with_parsed_rule(Params = #{id := RuleId, sql := Sql, actions := Actions}, CreatedAt, Fun) ->
     case emqx_rule_sqlparser:parse(Sql) of
         {ok, Select} ->
             Rule = #{
@@ -459,7 +477,7 @@ parse_and_insert(Params = #{id := RuleId, sql := Sql, actions := Actions}, Creat
                 conditions => emqx_rule_sqlparser:select_where(Select)
                 %% -- calculated fields end
             },
-            ok = insert_rule(Rule),
+            ok = Fun(Rule),
             {ok, Rule};
         {error, Reason} ->
             {error, Reason}
@@ -471,16 +489,27 @@ do_insert_rule(#{id := Id} = Rule) ->
     true = ets:insert(?RULE_TAB, {Id, maps:remove(id, Rule)}),
     ok.
 
-do_delete_rule(RuleId) ->
-    case get_rule(RuleId) of
-        {ok, Rule} ->
-            ok = unload_hooks_for_rule(Rule),
-            ok = clear_metrics_for_rule(RuleId),
-            true = ets:delete(?RULE_TAB, RuleId),
-            ok;
-        not_found ->
-            ok
-    end.
+do_delete_rule(#{id := Id} = Rule) ->
+    ok = unload_hooks_for_rule(Rule),
+    ok = clear_metrics_for_rule(Id),
+    true = ets:delete(?RULE_TAB, Id),
+    ok.
+
+do_update_rule_index(#{id := Id, from := From} = Rule) ->
+    ok = lists:foreach(
+        fun(Topic) ->
+            true = emqx_rule_index:insert(Topic, Id, Rule, ?RULE_TOPIC_INDEX)
+        end,
+        From
+    ).
+
+do_delete_rule_index(#{id := Id, from := From}) ->
+    ok = lists:foreach(
+        fun(Topic) ->
+            true = emqx_rule_index:delete(Topic, Id, ?RULE_TOPIC_INDEX)
+        end,
+        From
+    ).
 
 parse_actions(Actions) ->
     [do_parse_action(Act) || Act <- Actions].

apps/emqx_rule_engine/src/emqx_rule_engine_app.erl

@@ -26,6 +26,7 @@
 
 start(_Type, _Args) ->
     _ = ets:new(?RULE_TAB, [named_table, public, ordered_set, {read_concurrency, true}]),
+    _ = ets:new(?RULE_TOPIC_INDEX, [named_table, public, ordered_set, {read_concurrency, true}]),
     ok = emqx_rule_events:reload(),
     SupRet = emqx_rule_engine_sup:start_link(),
     ok = emqx_rule_engine:load_rules(),

apps/emqx_rule_engine/src/emqx_rule_index.erl

@@ -0,0 +1,140 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. 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
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% 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.
+%%--------------------------------------------------------------------
+
+%% @doc Topic index for matching topics to topic filters.
+%%
+%% 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?
+%%
+%% 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/2]).
+
+-export([get_record/2]).
+
+-type key(ID) :: [binary() | '+' | '#' | {ID}].
+-type match(ID) :: key(ID).
+
+-ifdef(TEST).
+-export_type([match/1]).
+-endif.
+
+insert(Filter, ID, Record, Tab) ->
+    %% 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}]).
+
+-spec match(emqx_types:topic(), ets:table()) -> match(_ID) | false.
+match(Topic, Tab) ->
+    {Words, RPrefix} = match_init(Topic),
+    match(Words, RPrefix, Tab).
+
+match(Words, RPrefix, Tab) ->
+    Prefix = lists:reverse(RPrefix),
+    K = ets:next(Tab, Prefix),
+    case match_filter(Prefix, K, Words =/= []) of
+        true ->
+            K;
+        stop ->
+            false;
+        Matched ->
+            match_rest(Matched, Words, RPrefix, Tab)
+    end.
+
+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 when is_list(Match) ->
+            Match;
+        false ->
+            match(Rest, [W | RPrefix], Tab)
+    end;
+match_rest(_, [], _RPrefix, _Tab) ->
+    false.
+
+-spec matches(emqx_types:topic(), ets:table()) -> [match(_ID)].
+matches(Topic, Tab) ->
+    {Words, RPrefix} = match_init(Topic),
+    matches(Words, RPrefix, Tab).
+
+matches(Words, RPrefix, Tab) ->
+    Prefix = lists:reverse(RPrefix),
+    matches(ets:next(Tab, Prefix), Prefix, Words, RPrefix, Tab).
+
+matches(K, Prefix, Words, RPrefix, Tab) ->
+    case match_filter(Prefix, K, Words =/= []) of
+        true ->
+            [K | matches(ets:next(Tab, K), Prefix, Words, RPrefix, Tab)];
+        stop ->
+            [];
+        Matched ->
+            matches_rest(Matched, Words, RPrefix, Tab)
+    end.
+
+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_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([], ['+' | _], _) ->
+    plus;
+match_filter([], [_H | _], _) ->
+    false;
+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) ->
+    case emqx_topic:words(Topic) of
+        [W = <<"$", _/bytes>> | Rest] ->
+            % NOTE
+            % This will effectively skip attempts to match special topics to `#` or `+/...`.
+            {Rest, [W]};
+        Words ->
+            {Words, []}
+    end.
+
+-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_index_SUITE.erl

@@ -0,0 +1,166 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. 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
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% 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.
+%%--------------------------------------------------------------------
+
+-module(emqx_rule_index_SUITE).
+
+-compile(export_all).
+-compile(nowarn_export_all).
+
+-include_lib("eunit/include/eunit.hrl").
+
+all() ->
+    emqx_common_test_helpers:all(?MODULE).
+
+t_insert(_) ->
+    Tab = new(),
+    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/#">>, get_topic(match(<<"sensor">>, Tab))),
+    ?assertEqual(t_insert_3, get_id(match(<<"sensor">>, Tab))),
+    true = ets:delete(Tab).
+
+t_match(_) ->
+    Tab = new(),
+    true = emqx_rule_index:insert(<<"sensor/1/metric/2">>, t_match_1, <<>>, Tab),
+    true = emqx_rule_index:insert(<<"sensor/+/#">>, t_match_2, <<>>, Tab),
+    true = emqx_rule_index:insert(<<"sensor/#">>, t_match_3, <<>>, Tab),
+    ?assertMatch(
+        [<<"sensor/#">>, <<"sensor/+/#">>],
+        [get_topic(M) || M <- matches(<<"sensor/1">>, Tab)]
+    ),
+    true = ets:delete(Tab).
+
+t_match2(_) ->
+    Tab = new(),
+    true = emqx_rule_index:insert(<<"#">>, t_match2_1, <<>>, Tab),
+    true = emqx_rule_index:insert(<<"+/#">>, t_match2_2, <<>>, Tab),
+    true = emqx_rule_index:insert(<<"+/+/#">>, t_match2_3, <<>>, Tab),
+    ?assertEqual(
+        [<<"#">>, <<"+/#">>, <<"+/+/#">>],
+        [get_topic(M) || M <- matches(<<"a/b/c">>, Tab)]
+    ),
+    ?assertEqual(
+        false,
+        emqx_rule_index:match(<<"$SYS/broker/zenmq">>, Tab)
+    ),
+    true = ets:delete(Tab).
+
+t_match3(_) ->
+    Tab = new(),
+    Records = [
+        {<<"d/#">>, t_match3_1},
+        {<<"a/b/+">>, t_match3_2},
+        {<<"a/#">>, t_match3_3},
+        {<<"#">>, t_match3_4},
+        {<<"$SYS/#">>, t_match3_sys}
+    ],
+    lists:foreach(
+        fun({Topic, ID}) -> emqx_rule_index:insert(Topic, ID, <<>>, Tab) end,
+        Records
+    ),
+    Matched = matches(<<"a/b/c">>, Tab),
+    case length(Matched) of
+        3 -> ok;
+        _ -> error({unexpected, Matched})
+    end,
+    ?assertEqual(
+        t_match3_sys,
+        get_id(match(<<"$SYS/a/b/c">>, Tab))
+    ),
+    true = ets:delete(Tab).
+
+t_match4(_) ->
+    Tab = new(),
+    Records = [{<<"/#">>, t_match4_1}, {<<"/+">>, t_match4_2}, {<<"/+/a/b/c">>, t_match4_3}],
+    lists:foreach(
+        fun({Topic, ID}) -> emqx_rule_index:insert(Topic, ID, <<>>, Tab) end,
+        Records
+    ),
+    ?assertEqual(
+        [<<"/#">>, <<"/+">>],
+        [get_topic(M) || M <- matches(<<"/">>, Tab)]
+    ),
+    ?assertEqual(
+        [<<"/#">>, <<"/+/a/b/c">>],
+        [get_topic(M) || M <- matches(<<"/0/a/b/c">>, Tab)]
+    ),
+    true = ets:delete(Tab).
+
+t_match5(_) ->
+    Tab = new(),
+    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">>,
+    Records = [
+        {<<"#">>, t_match5_1},
+        {<<T/binary, "/#">>, t_match5_2},
+        {<<T/binary, "/+">>, t_match5_3}
+    ],
+    lists:foreach(
+        fun({Topic, ID}) -> emqx_rule_index:insert(Topic, ID, <<>>, Tab) end,
+        Records
+    ),
+    ?assertEqual(
+        [<<"#">>, <<T/binary, "/#">>],
+        [get_topic(M) || M <- matches(T, Tab)]
+    ),
+    ?assertEqual(
+        [<<"#">>, <<T/binary, "/#">>, <<T/binary, "/+">>],
+        [get_topic(M) || M <- matches(<<T/binary, "/1">>, Tab)]
+    ),
+    true = ets:delete(Tab).
+
+t_match6(_) ->
+    Tab = new(),
+    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, get_id(match(T, Tab))),
+    true = ets:delete(Tab).
+
+t_match7(_) ->
+    Tab = new(),
+    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, get_topic(match(T, Tab))),
+    true = ets:delete(Tab).
+
+new() ->
+    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)).
+
+-spec get_id(emqx_rule_index:match(ID)) -> ID.
+get_id([{ID}]) ->
+    ID;
+get_id([_ | Rest]) ->
+    get_id(Rest).
+
+-spec get_topic(emqx_rule_index:match(_ID)) -> emqx_types:topic().
+get_topic(K) ->
+    emqx_topic:join(cut_topic(K)).
+
+cut_topic(K) ->
+    cut_topic(K, []).
+
+cut_topic([{_ID}], Acc) ->
+    lists:reverse(Acc);
+cut_topic([W | Rest], Acc) ->
+    cut_topic(Rest, [W | Acc]).

changes/ce/perf-11282.en.md

@@ -0,0 +1 @@
+Added indexing to the rule engine's topic matching to improve rule search performance.