Use a graph-like projection for topic bindings in Khepri

Performance is better than mnesia: 72k msg/s vs. 55k msg/s on my
machine with this PerfTest.

    perf-test -p -e amq.topic -t topic -qp q.%d -qpf 1 -qpt 2 -c 3000

This is for two reasons:

* In general, mnesia calls have some overhead for looking up the table
  and calling internal routines. Eventually the ETS calls end up being
  roughly the same as pure ETS equivalents. This is already noted in
  some places in the rabbit codebase.
* The mnesia version of this `trie_match/2` function occurs inside an
  `mnesia:async_dirty/2` call - inside a transaction. This has a lot
  more overhead from locking the table which is eliminated when
  switching to pure-ETS calls. I don't think the transaction has any
  practical importance on the results of `trie_match/2` here. Some
  races between updates and queries are possible but I think highly
  unlikely.

This approach should grow fairly well and be tolerant to binding churn:
the projection of edges is a set so it should handle insertion/
deletions/updates of bindings well.

This replaces a version of topic routing that uses regular expressions
to perform the matching. Regular expressions are much faster at
confirming/denying that a routing key matches a pattern but I suspect
that it won't grow very well as the number of topic bindings increases
because you need to do a full (or at least partial) table scan to find
all of the bindings for the exchange first. If there are many bindings,
I suspect that the lookup of bindings for an exchange will dominate the
running time.

TODO:

* Figure out exactly where this code should belong. It's awkwardly
  split between the projection definition and
  `rabbit_db_topic_exchange` for now.
* Check the correctness of the graph approach.
* Compare the performance of the graph approach to the regex approach.

2 files changed, 132 insertions, 70 deletions

diff --git a/deps/rabbit/src/rabbit_db_topic_exchange.erl b/deps/rabbit/src/rabbit_db_topic_exchange.erl
index 9cb730fe37..5b9b78b8ee 100644
--- a/deps/rabbit/src/rabbit_db_topic_exchange.erl
+++ b/deps/rabbit/src/rabbit_db_topic_exchange.erl
@@ -37,13 +37,6 @@
 -define(MNESIA_EDGE_TABLE, rabbit_topic_trie_edge).
 -define(MNESIA_BINDING_TABLE, rabbit_topic_trie_binding).
 
--define(HASH, <<"#">>).
--define(STAR, <<"*">>).
--define(DOT, <<"\\.">>).
--define(ONE_WORD, <<"[^.]+">>).
--define(ANYTHING, <<".*">>).
--define(ZERO_OR_MORE, <<"(\\..+)?">>).
-
 %% -------------------------------------------------------------------
 %% set().
 %% -------------------------------------------------------------------
@@ -197,23 +190,8 @@ match_in_mnesia(XName, RoutingKey) ->
     mnesia:async_dirty(fun trie_match/2, [XName, Words]).
 
 match_in_khepri(XName, RoutingKey) ->
-    Root = khepri_exchange_type_topic_path(XName) ++ [rabbit_khepri:if_has_data_wildcard()],
-    case rabbit_khepri:fold(
-           Root,
-           fun(Path0, #{data := Set}, Acc) ->
-                   Path = lists:nthtail(4, Path0),
-                   case is_re_topic_match(Path, RoutingKey) of
-                       true ->
-                           Bindings = sets:to_list(Set),
-                           [maps:get(destination, B) || B <- Bindings] ++ Acc;
-                       false ->
-                           Acc
-                   end
-           end,
-           []) of
-        {ok, List} -> List;
-        _ -> []
-    end.
+    Words = split_topic_key_binary(RoutingKey),
+    trie_match_in_khepri(XName, Words).
 
 %% --------------------------------------------------------------
 %% Migration
@@ -590,51 +568,6 @@ split_topic_key_binary(Key) ->
     Words = split_topic_key(Key, [], []),
     [list_to_binary(W) || W <- Words].
 
-is_re_topic_match([?HASH], _) ->
-    true;
-is_re_topic_match([A], A) ->
-    true;
-is_re_topic_match([], <<>>) ->
-    true;
-is_re_topic_match([], _) ->
-    false;
-is_re_topic_match(Path00, RoutingKey) ->
-    Path0 = path_to_re(Path00),
-    Path = << <<B/binary >> || B <- Path0 >>,
-    case Path of
-        ?ANYTHING -> true;
-        _ ->
-            RE = <<$^,Path/binary,$$>>,
-            case re:run(RoutingKey, RE, [{capture, none}]) of
-                nomatch -> false;
-                _ -> true
-            end
-    end.
-
-path_to_re([?STAR | Rest]) ->
-    path_to_re(Rest, [?ONE_WORD]);
-path_to_re([?HASH | Rest]) ->
-    path_to_re(Rest, [?ANYTHING]);
-path_to_re([Bin | Rest]) ->
-    path_to_re(Rest, [Bin]).
-
-path_to_re([], Acc) ->
-    lists:reverse(Acc);
-path_to_re([?STAR | Rest], [?ANYTHING | _] = Acc) ->
-    path_to_re(Rest, [?ONE_WORD | Acc]);
-path_to_re([?STAR | Rest], Acc) ->
-    path_to_re(Rest, [?ONE_WORD, ?DOT | Acc]);
-path_to_re([?HASH | Rest], [?HASH | _] = Acc) ->
-    path_to_re(Rest, Acc);
-path_to_re([?HASH | Rest], [?ANYTHING | _] = Acc) ->
-    path_to_re(Rest, Acc);
-path_to_re([?HASH | Rest], Acc) ->
-    path_to_re(Rest, [?ZERO_OR_MORE | Acc]);
-path_to_re([Bin | Rest], [?ANYTHING | _] = Acc) ->
-    path_to_re(Rest, [Bin | Acc]);
-path_to_re([Bin | Rest], Acc) ->
-    path_to_re(Rest, [Bin, ?DOT | Acc]).
-
 ensure_topic_deletion_ets() ->
     Tab = rabbit_db_topic_exchange_delete_table,
     case ets:whereis(Tab) of
@@ -652,3 +585,55 @@ ensure_topic_migration_ets() ->
         Tid ->
             Tid
     end.
+
+%% Khepri topic graph
+
+trie_match_in_khepri(X, Words) ->
+    trie_match_in_khepri(X, root, Words, []).
+
+trie_match_in_khepri(X, Node, [], ResAcc) ->
+    trie_match_part_in_khepri(
+      X, Node, <<"#">>, fun trie_match_skip_any_in_khepri/4, [],
+      trie_bindings_in_khepri(X, Node) ++ ResAcc);
+trie_match_in_khepri(X, Node, [W | RestW] = Words, ResAcc) ->
+    lists:foldl(fun ({WArg, MatchFun, RestWArg}, Acc) ->
+                        trie_match_part_in_khepri(
+                          X, Node, WArg, MatchFun, RestWArg, Acc)
+                end, ResAcc, [{W, fun trie_match_in_khepri/4, RestW},
+                              {<<"*">>, fun trie_match_in_khepri/4, RestW},
+                              {<<"#">>,
+                               fun trie_match_skip_any_in_khepri/4, Words}]).
+
+trie_match_part_in_khepri(X, Node, Search, MatchFun, RestW, ResAcc) ->
+    case trie_child_in_khepri(X, Node, Search) of
+        {ok, NextNode} -> MatchFun(X, NextNode, RestW, ResAcc);
+        error          -> ResAcc
+    end.
+
+trie_match_skip_any_in_khepri(X, Node, [], ResAcc) ->
+    trie_match_in_khepri(X, Node, [], ResAcc);
+trie_match_skip_any_in_khepri(X, Node, [_ | RestW] = Words, ResAcc) ->
+    trie_match_skip_any_in_khepri(
+      X, Node, RestW,
+      trie_match_in_khepri(X, Node, Words, ResAcc)).
+
+trie_child_in_khepri(X, Node, Word) ->
+    case ets:lookup(rabbit_khepri_topic_trie,
+                    #trie_edge{exchange_name = X,
+                               node_id       = Node,
+                               word          = Word}) of
+        [#topic_trie_edge{node_id = NextNode}] -> {ok, NextNode};
+        []                                     -> error
+    end.
+
+trie_bindings_in_khepri(X, Node) ->
+    case ets:lookup(rabbit_khepri_topic_trie,
+                    #trie_edge{exchange_name = X,
+                               node_id       = Node,
+                               word          = bindings}) of
+        [#topic_trie_edge{node_id = {bindings, Bindings}}] ->
+            [Dest || #{destination := Dest} <- sets:to_list(Bindings)];
+        [] ->
+            []
+    end.
+
diff --git a/deps/rabbit/src/rabbit_khepri.erl b/deps/rabbit/src/rabbit_khepri.erl
index c85aae9828..a3d3f1eb51 100644
--- a/deps/rabbit/src/rabbit_khepri.erl
+++ b/deps/rabbit/src/rabbit_khepri.erl
@@ -596,7 +596,8 @@ register_projections() ->
                     fun register_rabbit_users_projection/0,
                     fun register_rabbit_user_permissions_projection/0,
                     fun register_rabbit_bindings_projection/0,
-                    fun register_rabbit_index_route_projection/0],
+                    fun register_rabbit_index_route_projection/0,
+                    fun register_rabbit_topic_graph_projection/0],
     [case RegisterFun() of
          ok              -> ok;
          {error, exists} -> ok;
@@ -719,3 +720,79 @@ projection_fun_for_sets(MapFun) ->
         (_Table, _Path, _OldProps, _NewProps) ->
             ok
     end.
+
+register_rabbit_topic_graph_projection() ->
+    Name = rabbit_khepri_topic_trie,
+    Options = #{keypos => #topic_trie_edge.trie_edge},
+    ProjectionFun =
+    fun (Table, Path, #{data := _OldBindings}, #{data := NewBindings}) ->
+            [BindingEdge | _RestEdges] = edges_for_path(Path, NewBindings),
+            ets:insert(Table, BindingEdge);
+        (Table, Path, _OldProps, #{data := NewBindings}) ->
+            Edges = edges_for_path(Path, NewBindings),
+            ets:insert(Table, Edges);
+        (Table, Path, #{data := OldBindings}, _NewProps) ->
+            [BindingEdge | RestEdges] = edges_for_path(Path, OldBindings),
+            ets:delete_object(Table, BindingEdge),
+            trim_while_out_degree_is_zero(RestEdges);
+        (_Table, _Path, _OldProps, _NewProps) ->
+            ok
+    end,
+    Projection = khepri_projection:new(Name, ProjectionFun, Options),
+    PathPattern = [rabbit_db_topic_exchange,
+                   topic_trie_binding,
+                   _VHost = ?KHEPRI_WILDCARD_STAR,
+                   _ExchangeName = ?KHEPRI_WILDCARD_STAR,
+                   _Routes = ?KHEPRI_WILDCARD_STAR_STAR],
+    khepri:register_projection(?RA_CLUSTER_NAME, PathPattern, Projection).
+
+edges_for_path(
+  [rabbit_db_topic_exchange, topic_trie_binding,
+   VHost, ExchangeName | Components],
+  Bindings) ->
+    Exchange = rabbit_misc:r(VHost, exchange, ExchangeName),
+    edges_for_path([root | Components], Bindings, Exchange, []).
+
+edges_for_path([FromNodeId, To | Rest], Bindings, Exchange, Edges) ->
+    ToNodeId = [To | FromNodeId],
+    Edge = #topic_trie_edge{trie_edge = #trie_edge{exchange_name = Exchange,
+                                                   node_id =       FromNodeId,
+                                                   word =          To},
+                            node_id = ToNodeId},
+    edges_for_path([ToNodeId | Rest], Bindings, Exchange, [Edge | Edges]);
+edges_for_path([LeafId], Bindings, Exchange, Edges) ->
+    ToNodeId = {bindings, Bindings},
+    Edge = #topic_trie_edge{trie_edge = #trie_edge{exchange_name = Exchange,
+                                                   node_id =       LeafId,
+                                                   word =          bindings},
+                            node_id = ToNodeId},
+    [Edge | Edges].
+
+-spec trim_while_out_degree_is_zero(Edges) -> ok
+    when
+      Edges :: [Edge],
+      Edge :: #topic_trie_edge{}.
+
+trim_while_out_degree_is_zero([]) ->
+    ok;
+trim_while_out_degree_is_zero([Edge | Rest]) ->
+    #topic_trie_edge{trie_edge = #trie_edge{exchange_name = Exchange,
+                                            node_id       = _FromNodeId},
+                     node_id = ToNodeId} = Edge,
+    OutEdgePattern = #topic_trie_edge{trie_edge =
+                                      #trie_edge{exchange_name = Exchange,
+                                                 node_id       = ToNodeId,
+                                                 word          = '_'},
+                                      node_id = '_'},
+    case ets:match(rabbit_khepri_topic_trie, OutEdgePattern, 1) of
+        '$end_of_table' ->
+            %% If the ToNode has an out degree of zero, trim the edge to
+            %% the node, effectively erasing ToNode.
+            ets:delete_object(rabbit_khepri_topic_trie, Edge),
+            trim_while_out_degree_is_zero(Rest);
+        {_Match, _Continuation} ->
+            %% Return after finding the first node with a non-zero out-degree.
+            %% If a node has a non-zero out-degree then all of its ancestors
+            %% must as well.
+            ok
+    end.