diff --git a/apps/emqx/include/emqx_release.hrl b/apps/emqx/include/emqx_release.hrl
index c7fa97be7..7b0fcad1a 100644
--- a/apps/emqx/include/emqx_release.hrl
+++ b/apps/emqx/include/emqx_release.hrl
@@ -35,7 +35,7 @@
 -define(EMQX_RELEASE_CE, "5.1.5-build.3").
 
 %% Enterprise edition
--define(EMQX_RELEASE_EE, "5.2.0-alpha.3").
+-define(EMQX_RELEASE_EE, "5.2.0-alpha.4").
 
 %% The HTTP API version
 -define(EMQX_API_VERSION, "5.0").
diff --git a/apps/emqx/include/emqx_router.hrl b/apps/emqx/include/emqx_router.hrl
index 035ff5455..35a267aa7 100644
--- a/apps/emqx/include/emqx_router.hrl
+++ b/apps/emqx/include/emqx_router.hrl
@@ -17,8 +17,9 @@
 -ifndef(EMQX_ROUTER_HRL).
 -define(EMQX_ROUTER_HRL, true).
 
-%% ETS table for message routing
+%% ETS tables for message routing
 -define(ROUTE_TAB, emqx_route).
+-define(ROUTE_TAB_FILTERS, emqx_route_filters).
 
 %% Mnesia table for message routing
 -define(ROUTING_NODE, emqx_routing_node).
diff --git a/apps/emqx/src/emqx_app.erl b/apps/emqx/src/emqx_app.erl
index 59a397836..0f4987085 100644
--- a/apps/emqx/src/emqx_app.erl
+++ b/apps/emqx/src/emqx_app.erl
@@ -55,7 +55,9 @@ prep_stop(_State) ->
     emqx_boot:is_enabled(listeners) andalso
         emqx_listeners:stop().
 
-stop(_State) -> ok.
+stop(_State) ->
+    ok = emqx_router:deinit_schema(),
+    ok.
 
 -define(CONFIG_LOADER, config_loader).
 -define(DEFAULT_LOADER, emqx).
diff --git a/apps/emqx/src/emqx_router.erl b/apps/emqx/src/emqx_router.erl
index 95b6136a7..464852ceb 100644
--- a/apps/emqx/src/emqx_router.erl
+++ b/apps/emqx/src/emqx_router.erl
@@ -21,7 +21,6 @@
 -include("emqx.hrl").
 -include("logger.hrl").
 -include("types.hrl").
--include_lib("mria/include/mria.hrl").
 -include_lib("emqx/include/emqx_router.hrl").
 
 %% Mnesia bootstrap
@@ -46,16 +45,25 @@
     do_delete_route/2
 ]).
 
+-export([cleanup_routes/1]).
+
 -export([
     match_routes/1,
-    lookup_routes/1,
-    has_routes/1
+    lookup_routes/1
 ]).
 
 -export([print_routes/1]).
 
+-export([
+    foldl_routes/2,
+    foldr_routes/2
+]).
+
 -export([topics/0]).
 
+%% Exported for tests
+-export([has_route/2]).
+
 %% gen_server callbacks
 -export([
     init/1,
@@ -66,10 +74,21 @@
     code_change/3
 ]).
 
+-export([
+    get_schema_vsn/0,
+    init_schema/0,
+    deinit_schema/0
+]).
+
 -type group() :: binary().
 
 -type dest() :: node() | {group(), node()}.
 
+-record(routeidx, {
+    entry :: emqx_topic_index:key(dest()),
+    unused = [] :: nil()
+}).
+
 %%--------------------------------------------------------------------
 %% Mnesia bootstrap
 %%--------------------------------------------------------------------
@@ -88,6 +107,19 @@ mnesia(boot) ->
                 {write_concurrency, true}
             ]}
         ]}
+    ]),
+    ok = mria:create_table(?ROUTE_TAB_FILTERS, [
+        {type, ordered_set},
+        {rlog_shard, ?ROUTE_SHARD},
+        {storage, ram_copies},
+        {record_name, routeidx},
+        {attributes, record_info(fields, routeidx)},
+        {storage_properties, [
+            {ets, [
+                {read_concurrency, true},
+                {write_concurrency, auto}
+            ]}
+        ]}
     ]).
 
 %%--------------------------------------------------------------------
@@ -121,43 +153,49 @@ do_add_route(Topic) when is_binary(Topic) ->
 
 -spec do_add_route(emqx_types:topic(), dest()) -> ok | {error, term()}.
 do_add_route(Topic, Dest) when is_binary(Topic) ->
-    Route = #route{topic = Topic, dest = Dest},
-    case lists:member(Route, lookup_routes(Topic)) of
+    case has_route(Topic, Dest) of
         true ->
             ok;
         false ->
             ok = emqx_router_helper:monitor(Dest),
-            case emqx_topic:wildcard(Topic) of
-                true ->
-                    Fun = fun emqx_router_utils:insert_trie_route/2,
-                    emqx_router_utils:maybe_trans(Fun, [?ROUTE_TAB, Route], ?ROUTE_SHARD);
-                false ->
-                    emqx_router_utils:insert_direct_route(?ROUTE_TAB, Route)
-            end
+            mria_insert_route(get_schema_vsn(), Topic, Dest)
     end.
 
-%% @doc Match routes
+mria_insert_route(v2, Topic, Dest) ->
+    mria_insert_route_v2(Topic, Dest);
+mria_insert_route(v1, Topic, Dest) ->
+    mria_insert_route_v1(Topic, Dest).
+
+%% @doc Take a real topic (not filter) as input, return the matching topics and topic
+%% filters associated with route destination.
 -spec match_routes(emqx_types:topic()) -> [emqx_types:route()].
 match_routes(Topic) when is_binary(Topic) ->
-    case match_trie(Topic) of
-        [] -> lookup_routes(Topic);
-        Matched -> lists:append([lookup_routes(To) || To <- [Topic | Matched]])
-    end.
+    match_routes(get_schema_vsn(), Topic).
 
-%% Optimize: routing table will be replicated to all router nodes.
-match_trie(Topic) ->
-    case emqx_trie:empty() of
-        true -> [];
-        false -> emqx_trie:match(Topic)
-    end.
+match_routes(v2, Topic) ->
+    match_routes_v2(Topic);
+match_routes(v1, Topic) ->
+    match_routes_v1(Topic).
 
+%% @doc Take a topic or filter as input, and return the existing routes with exactly
+%% this topic or filter.
 -spec lookup_routes(emqx_types:topic()) -> [emqx_types:route()].
 lookup_routes(Topic) ->
-    ets:lookup(?ROUTE_TAB, Topic).
+    lookup_routes(get_schema_vsn(), Topic).
 
--spec has_routes(emqx_types:topic()) -> boolean().
-has_routes(Topic) when is_binary(Topic) ->
-    ets:member(?ROUTE_TAB, Topic).
+lookup_routes(v2, Topic) ->
+    lookup_routes_v2(Topic);
+lookup_routes(v1, Topic) ->
+    lookup_routes_v1(Topic).
+
+-spec has_route(emqx_types:topic(), dest()) -> boolean().
+has_route(Topic, Dest) ->
+    has_route(get_schema_vsn(), Topic, Dest).
+
+has_route(v2, Topic, Dest) ->
+    has_route_v2(Topic, Dest);
+has_route(v1, Topic, Dest) ->
+    has_route_v1(Topic, Dest).
 
 -spec delete_route(emqx_types:topic()) -> ok | {error, term()}.
 delete_route(Topic) when is_binary(Topic) ->
@@ -173,18 +211,21 @@ do_delete_route(Topic) when is_binary(Topic) ->
 
 -spec do_delete_route(emqx_types:topic(), dest()) -> ok | {error, term()}.
 do_delete_route(Topic, Dest) ->
-    Route = #route{topic = Topic, dest = Dest},
-    case emqx_topic:wildcard(Topic) of
-        true ->
-            Fun = fun emqx_router_utils:delete_trie_route/2,
-            emqx_router_utils:maybe_trans(Fun, [?ROUTE_TAB, Route], ?ROUTE_SHARD);
-        false ->
-            emqx_router_utils:delete_direct_route(?ROUTE_TAB, Route)
-    end.
+    mria_delete_route(get_schema_vsn(), Topic, Dest).
+
+mria_delete_route(v2, Topic, Dest) ->
+    mria_delete_route_v2(Topic, Dest);
+mria_delete_route(v1, Topic, Dest) ->
+    mria_delete_route_v1(Topic, Dest).
 
 -spec topics() -> list(emqx_types:topic()).
 topics() ->
-    mnesia:dirty_all_keys(?ROUTE_TAB).
+    topics(get_schema_vsn()).
+
+topics(v2) ->
+    list_topics_v2();
+topics(v1) ->
+    list_topics_v1().
 
 %% @doc Print routes to a topic
 -spec print_routes(emqx_types:topic()) -> ok.
@@ -196,12 +237,290 @@ print_routes(Topic) ->
         match_routes(Topic)
     ).
 
+-spec cleanup_routes(node()) -> ok.
+cleanup_routes(Node) ->
+    cleanup_routes(get_schema_vsn(), Node).
+
+cleanup_routes(v2, Node) ->
+    cleanup_routes_v2(Node);
+cleanup_routes(v1, Node) ->
+    cleanup_routes_v1(Node).
+
+-spec foldl_routes(fun((emqx_types:route(), Acc) -> Acc), Acc) -> Acc.
+foldl_routes(FoldFun, AccIn) ->
+    fold_routes(get_schema_vsn(), foldl, FoldFun, AccIn).
+
+-spec foldr_routes(fun((emqx_types:route(), Acc) -> Acc), Acc) -> Acc.
+foldr_routes(FoldFun, AccIn) ->
+    fold_routes(get_schema_vsn(), foldr, FoldFun, AccIn).
+
+fold_routes(v2, FunName, FoldFun, AccIn) ->
+    fold_routes_v2(FunName, FoldFun, AccIn);
+fold_routes(v1, FunName, FoldFun, AccIn) ->
+    fold_routes_v1(FunName, FoldFun, AccIn).
+
 call(Router, Msg) ->
     gen_server:call(Router, Msg, infinity).
 
 pick(Topic) ->
     gproc_pool:pick_worker(router_pool, Topic).
 
+%%--------------------------------------------------------------------
+%% Schema v1
+%% --------------------------------------------------------------------
+
+-dialyzer({nowarn_function, [cleanup_routes_v1/1]}).
+
+mria_insert_route_v1(Topic, Dest) ->
+    Route = #route{topic = Topic, dest = Dest},
+    case emqx_topic:wildcard(Topic) of
+        true ->
+            mria_route_tab_insert_update_trie(Route);
+        false ->
+            mria_route_tab_insert(Route)
+    end.
+
+mria_route_tab_insert_update_trie(Route) ->
+    emqx_router_utils:maybe_trans(
+        fun emqx_router_utils:insert_trie_route/2,
+        [?ROUTE_TAB, Route],
+        ?ROUTE_SHARD
+    ).
+
+mria_route_tab_insert(Route) ->
+    mria:dirty_write(?ROUTE_TAB, Route).
+
+mria_delete_route_v1(Topic, Dest) ->
+    Route = #route{topic = Topic, dest = Dest},
+    case emqx_topic:wildcard(Topic) of
+        true ->
+            mria_route_tab_delete_update_trie(Route);
+        false ->
+            mria_route_tab_delete(Route)
+    end.
+
+mria_route_tab_delete_update_trie(Route) ->
+    emqx_router_utils:maybe_trans(
+        fun emqx_router_utils:delete_trie_route/2,
+        [?ROUTE_TAB, Route],
+        ?ROUTE_SHARD
+    ).
+
+mria_route_tab_delete(Route) ->
+    mria:dirty_delete_object(?ROUTE_TAB, Route).
+
+match_routes_v1(Topic) ->
+    lookup_route_tab(Topic) ++
+        lists:flatmap(fun lookup_route_tab/1, match_global_trie(Topic)).
+
+match_global_trie(Topic) ->
+    case emqx_trie:empty() of
+        true -> [];
+        false -> emqx_trie:match(Topic)
+    end.
+
+lookup_routes_v1(Topic) ->
+    lookup_route_tab(Topic).
+
+lookup_route_tab(Topic) ->
+    ets:lookup(?ROUTE_TAB, Topic).
+
+has_route_v1(Topic, Dest) ->
+    has_route_tab_entry(Topic, Dest).
+
+has_route_tab_entry(Topic, Dest) ->
+    [] =/= ets:match(?ROUTE_TAB, #route{topic = Topic, dest = Dest}).
+
+cleanup_routes_v1(Node) ->
+    Patterns = [
+        #route{_ = '_', dest = Node},
+        #route{_ = '_', dest = {'_', Node}}
+    ],
+    mria:transaction(?ROUTE_SHARD, fun() ->
+        [
+            mnesia:delete_object(?ROUTE_TAB, Route, write)
+         || Pat <- Patterns,
+            Route <- mnesia:match_object(?ROUTE_TAB, Pat, write)
+        ]
+    end).
+
+list_topics_v1() ->
+    list_route_tab_topics().
+
+list_route_tab_topics() ->
+    mnesia:dirty_all_keys(?ROUTE_TAB).
+
+fold_routes_v1(FunName, FoldFun, AccIn) ->
+    ets:FunName(FoldFun, AccIn, ?ROUTE_TAB).
+
+%%--------------------------------------------------------------------
+%% Schema v2
+%% One bag table exclusively for regular, non-filter subscription
+%% topics, and one `emqx_topic_index` table exclusively for wildcard
+%% topics. Writes go to only one of the two tables at a time.
+%% --------------------------------------------------------------------
+
+mria_insert_route_v2(Topic, Dest) ->
+    case emqx_trie_search:filter(Topic) of
+        Words when is_list(Words) ->
+            K = emqx_topic_index:make_key(Words, Dest),
+            mria:dirty_write(?ROUTE_TAB_FILTERS, #routeidx{entry = K});
+        false ->
+            mria_route_tab_insert(#route{topic = Topic, dest = Dest})
+    end.
+
+mria_delete_route_v2(Topic, Dest) ->
+    case emqx_trie_search:filter(Topic) of
+        Words when is_list(Words) ->
+            K = emqx_topic_index:make_key(Words, Dest),
+            mria:dirty_delete(?ROUTE_TAB_FILTERS, K);
+        false ->
+            mria_route_tab_delete(#route{topic = Topic, dest = Dest})
+    end.
+
+match_routes_v2(Topic) ->
+    lookup_route_tab(Topic) ++
+        [match_to_route(M) || M <- match_filters(Topic)].
+
+match_filters(Topic) ->
+    emqx_topic_index:matches(Topic, ?ROUTE_TAB_FILTERS, []).
+
+lookup_routes_v2(Topic) ->
+    case emqx_topic:wildcard(Topic) of
+        true ->
+            Pat = #routeidx{entry = emqx_topic_index:make_key(Topic, '$1')},
+            [Dest || [Dest] <- ets:match(?ROUTE_TAB_FILTERS, Pat)];
+        false ->
+            lookup_route_tab(Topic)
+    end.
+
+has_route_v2(Topic, Dest) ->
+    case emqx_topic:wildcard(Topic) of
+        true ->
+            ets:member(?ROUTE_TAB_FILTERS, emqx_topic_index:make_key(Topic, Dest));
+        false ->
+            has_route_tab_entry(Topic, Dest)
+    end.
+
+cleanup_routes_v2(Node) ->
+    % NOTE
+    % No point in transaction here because all the operations on filters table are dirty.
+    ok = ets:foldl(
+        fun(#routeidx{entry = K}, ok) ->
+            case get_dest_node(emqx_topic_index:get_id(K)) of
+                Node ->
+                    mria:dirty_delete(?ROUTE_TAB_FILTERS, K);
+                _ ->
+                    ok
+            end
+        end,
+        ok,
+        ?ROUTE_TAB_FILTERS
+    ),
+    ok = ets:foldl(
+        fun(#route{dest = Dest} = Route, ok) ->
+            case get_dest_node(Dest) of
+                Node ->
+                    mria:dirty_delete_object(?ROUTE_TAB, Route);
+                _ ->
+                    ok
+            end
+        end,
+        ok,
+        ?ROUTE_TAB
+    ).
+
+get_dest_node({_, Node}) ->
+    Node;
+get_dest_node(Node) ->
+    Node.
+
+list_topics_v2() ->
+    Pat = #routeidx{entry = '$1'},
+    Filters = [emqx_topic_index:get_topic(K) || [K] <- ets:match(?ROUTE_TAB_FILTERS, Pat)],
+    list_route_tab_topics() ++ Filters.
+
+fold_routes_v2(FunName, FoldFun, AccIn) ->
+    FilterFoldFun = mk_filtertab_fold_fun(FoldFun),
+    Acc = ets:FunName(FoldFun, AccIn, ?ROUTE_TAB),
+    ets:FunName(FilterFoldFun, Acc, ?ROUTE_TAB_FILTERS).
+
+mk_filtertab_fold_fun(FoldFun) ->
+    fun(#routeidx{entry = K}, Acc) -> FoldFun(match_to_route(K), Acc) end.
+
+match_to_route(M) ->
+    #route{topic = emqx_topic_index:get_topic(M), dest = emqx_topic_index:get_id(M)}.
+
+%%--------------------------------------------------------------------
+%% Routing table type
+%% --------------------------------------------------------------------
+
+-define(PT_SCHEMA_VSN, {?MODULE, schemavsn}).
+
+-type schemavsn() :: v1 | v2.
+
+-spec get_schema_vsn() -> schemavsn().
+get_schema_vsn() ->
+    persistent_term:get(?PT_SCHEMA_VSN).
+
+-spec init_schema() -> ok.
+init_schema() ->
+    ok = mria:wait_for_tables([?ROUTE_TAB, ?ROUTE_TAB_FILTERS]),
+    ok = emqx_trie:wait_for_tables(),
+    ConfSchema = emqx_config:get([broker, routing, storage_schema]),
+    Schema = choose_schema_vsn(ConfSchema),
+    ok = persistent_term:put(?PT_SCHEMA_VSN, Schema),
+    case Schema of
+        ConfSchema ->
+            ?SLOG(info, #{
+                msg => "routing_schema_used",
+                schema => Schema
+            });
+        _ ->
+            ?SLOG(notice, #{
+                msg => "configured_routing_schema_ignored",
+                schema_in_use => Schema,
+                configured => ConfSchema,
+                reason =>
+                    "Could not use configured routing storage schema because "
+                    "there are already non-empty routing tables pertaining to "
+                    "another schema."
+            })
+    end.
+
+-spec deinit_schema() -> ok.
+deinit_schema() ->
+    _ = persistent_term:erase(?PT_SCHEMA_VSN),
+    ok.
+
+-spec choose_schema_vsn(schemavsn()) -> schemavsn().
+choose_schema_vsn(ConfType) ->
+    IsEmptyIndex = emqx_trie:empty(),
+    IsEmptyFilters = is_empty(?ROUTE_TAB_FILTERS),
+    case {IsEmptyIndex, IsEmptyFilters} of
+        {true, true} ->
+            ConfType;
+        {false, true} ->
+            v1;
+        {true, false} ->
+            v2;
+        {false, false} ->
+            ?SLOG(critical, #{
+                msg => "conflicting_routing_schemas_detected_in_cluster",
+                configured => ConfType,
+                reason =>
+                    "There are records in the routing tables related to both v1 "
+                    "and v2 storage schemas. This probably means that some nodes "
+                    "in the cluster use v1 schema and some use v2, independently "
+                    "of each other. The routing is likely broken. Manual intervention "
+                    "and full cluster restart is required. This node will shut down."
+            }),
+            error(conflicting_routing_schemas_detected_in_cluster)
+    end.
+
+is_empty(Tab) ->
+    ets:first(Tab) =:= '$end_of_table'.
+
 %%--------------------------------------------------------------------
 %% gen_server callbacks
 %%--------------------------------------------------------------------
diff --git a/apps/emqx/src/emqx_router_helper.erl b/apps/emqx/src/emqx_router_helper.erl
index 8d96bf81d..b9cdbae4b 100644
--- a/apps/emqx/src/emqx_router_helper.erl
+++ b/apps/emqx/src/emqx_router_helper.erl
@@ -148,11 +148,12 @@ handle_info({mnesia_table_event, Event}, State) ->
 handle_info({nodedown, Node}, State = #{nodes := Nodes}) ->
     case mria_rlog:role() of
         core ->
+            % TODO
+            % Node may flap, do we need to wait for any pending cleanups in `init/1`
+            % on the flapping node?
             global:trans(
                 {?LOCK, self()},
-                fun() ->
-                    mria:transaction(?ROUTE_SHARD, fun ?MODULE:cleanup_routes/1, [Node])
-                end
+                fun() -> cleanup_routes(Node) end
             ),
             ok = mria:dirty_delete(?ROUTING_NODE, Node);
         replicant ->
@@ -197,11 +198,4 @@ stats_fun() ->
     end.
 
 cleanup_routes(Node) ->
-    Patterns = [
-        #route{_ = '_', dest = Node},
-        #route{_ = '_', dest = {'_', Node}}
-    ],
-    [
-        mnesia:delete_object(?ROUTE_TAB, Route, write)
-     || Pat <- Patterns, Route <- mnesia:match_object(?ROUTE_TAB, Pat, write)
-    ].
+    emqx_router:cleanup_routes(Node).
diff --git a/apps/emqx/src/emqx_router_sup.erl b/apps/emqx/src/emqx_router_sup.erl
index 0fa48d9d2..588b0de8e 100644
--- a/apps/emqx/src/emqx_router_sup.erl
+++ b/apps/emqx/src/emqx_router_sup.erl
@@ -23,6 +23,8 @@
 -export([init/1]).
 
 start_link() ->
+    %% Init and log routing table type
+    ok = emqx_router:init_schema(),
     supervisor:start_link({local, ?MODULE}, ?MODULE, []).
 
 init([]) ->
diff --git a/apps/emqx/src/emqx_schema.erl b/apps/emqx/src/emqx_schema.erl
index db2c0b4a8..04bd397ec 100644
--- a/apps/emqx/src/emqx_schema.erl
+++ b/apps/emqx/src/emqx_schema.erl
@@ -1358,6 +1358,11 @@ fields("broker") ->
                 ref("broker_perf"),
                 #{importance => ?IMPORTANCE_HIDDEN}
             )},
+        {"routing",
+            sc(
+                ref("broker_routing"),
+                #{importance => ?IMPORTANCE_HIDDEN}
+            )},
         %% FIXME: Need new design for shared subscription group
         {"shared_subscription_group",
             sc(
@@ -1369,6 +1374,18 @@ fields("broker") ->
                 }
             )}
     ];
+fields("broker_routing") ->
+    [
+        {"storage_schema",
+            sc(
+                hoconsc:enum([v1, v2]),
+                #{
+                    default => v1,
+                    'readOnly' => true,
+                    desc => ?DESC(broker_routing_storage_schema)
+                }
+            )}
+    ];
 fields("shared_subscription_group") ->
     [
         {"strategy",
diff --git a/apps/emqx/src/emqx_topic_gbt.erl b/apps/emqx/src/emqx_topic_gbt.erl
index dade70cee..063cba21d 100644
--- a/apps/emqx/src/emqx_topic_gbt.erl
+++ b/apps/emqx/src/emqx_topic_gbt.erl
@@ -29,8 +29,8 @@
 -export([get_topic/1]).
 -export([get_record/2]).
 
--type word() :: binary() | '+' | '#'.
--type key(ID) :: {[word()], {ID}}.
+-type key(ID) :: emqx_trie_search:key(ID).
+-type words() :: emqx_trie_search:words().
 -type match(ID) :: key(ID).
 -type name() :: any().
 
@@ -50,7 +50,7 @@ new(Name) ->
 %% @doc Insert a new entry into the index that associates given topic filter to given
 %% record ID, and attaches arbitrary record to the entry. This allows users to choose
 %% between regular and "materialized" indexes, for example.
--spec insert(emqx_types:topic(), _ID, _Record, name()) -> true.
+-spec insert(emqx_types:topic() | words(), _ID, _Record, name()) -> true.
 insert(Filter, ID, Record, Name) ->
     Tree = gbt(Name),
     Key = key(Filter, ID),
@@ -59,7 +59,7 @@ insert(Filter, ID, Record, Name) ->
 
 %% @doc Delete an entry from the index that associates given topic filter to given
 %% record ID. Deleting non-existing entry is not an error.
--spec delete(emqx_types:topic(), _ID, name()) -> true.
+-spec delete(emqx_types:topic() | words(), _ID, name()) -> true.
 delete(Filter, ID, Name) ->
     Tree = gbt(Name),
     Key = key(Filter, ID),
diff --git a/apps/emqx/src/emqx_topic_index.erl b/apps/emqx/src/emqx_topic_index.erl
index acbf36627..59dfdfeab 100644
--- a/apps/emqx/src/emqx_topic_index.erl
+++ b/apps/emqx/src/emqx_topic_index.erl
@@ -24,12 +24,15 @@
 -export([match/2]).
 -export([matches/3]).
 
+-export([make_key/2]).
+
 -export([get_id/1]).
 -export([get_topic/1]).
 -export([get_record/2]).
 
 -type key(ID) :: emqx_trie_search:key(ID).
 -type match(ID) :: key(ID).
+-type words() :: emqx_trie_search:words().
 
 %% @doc Create a new ETS table suitable for topic index.
 %% Usable mostly for testing purposes.
@@ -40,16 +43,20 @@ new() ->
 %% @doc Insert a new entry into the index that associates given topic filter to given
 %% record ID, and attaches arbitrary record to the entry. This allows users to choose
 %% between regular and "materialized" indexes, for example.
--spec insert(emqx_types:topic(), _ID, _Record, ets:table()) -> true.
+-spec insert(emqx_types:topic() | words(), _ID, _Record, ets:table()) -> true.
 insert(Filter, ID, Record, Tab) ->
-    Key = key(Filter, ID),
+    Key = make_key(Filter, ID),
     true = ets:insert(Tab, {Key, Record}).
 
 %% @doc Delete an entry from the index that associates given topic filter to given
 %% record ID. Deleting non-existing entry is not an error.
--spec delete(emqx_types:topic(), _ID, ets:table()) -> true.
+-spec delete(emqx_types:topic() | words(), _ID, ets:table()) -> true.
 delete(Filter, ID, Tab) ->
-    true = ets:delete(Tab, key(Filter, ID)).
+    ets:delete(Tab, make_key(Filter, ID)).
+
+-spec make_key(emqx_types:topic() | words(), ID) -> key(ID).
+make_key(TopicOrFilter, ID) ->
+    emqx_trie_search:make_key(TopicOrFilter, ID).
 
 %% @doc Match given topic against the index and return the first match, or `false` if
 %% no match is found.
@@ -73,13 +80,16 @@ get_topic(Key) ->
     emqx_trie_search:get_topic(Key).
 
 %% @doc Fetch the record associated with the match.
-%% NOTE: Only really useful for ETS tables where the record ID is the first element.
--spec get_record(match(_ID), ets:table()) -> _Record.
+%% May return empty list if the index entry was deleted in the meantime.
+%% NOTE: Only really useful for ETS tables where the record data is the last element.
+-spec get_record(match(_ID), ets:table()) -> [_Record].
 get_record(K, Tab) ->
-    ets:lookup_element(Tab, K, 2).
-
-key(TopicOrFilter, ID) ->
-    emqx_trie_search:make_key(TopicOrFilter, ID).
+    case ets:lookup(Tab, K) of
+        [Entry] ->
+            [erlang:element(tuple_size(Entry), Entry)];
+        [] ->
+            []
+    end.
 
 make_nextf(Tab) ->
     fun(Key) -> ets:next(Tab, Key) end.
diff --git a/apps/emqx/src/emqx_trie.erl b/apps/emqx/src/emqx_trie.erl
index 229a0e3f4..76be97d3e 100644
--- a/apps/emqx/src/emqx_trie.erl
+++ b/apps/emqx/src/emqx_trie.erl
@@ -21,6 +21,7 @@
 %% Mnesia bootstrap
 -export([
     mnesia/1,
+    wait_for_tables/0,
     create_session_trie/1
 ]).
 
@@ -105,6 +106,10 @@ create_session_trie(Type) ->
         ]
     ).
 
+-spec wait_for_tables() -> ok | {error, _Reason}.
+wait_for_tables() ->
+    mria:wait_for_tables([?TRIE]).
+
 %%--------------------------------------------------------------------
 %% Topics APIs
 %%--------------------------------------------------------------------
diff --git a/apps/emqx/src/emqx_trie_search.erl b/apps/emqx/src/emqx_trie_search.erl
index b774e1459..c8c088b58 100644
--- a/apps/emqx/src/emqx_trie_search.erl
+++ b/apps/emqx/src/emqx_trie_search.erl
@@ -98,24 +98,24 @@
 
 -module(emqx_trie_search).
 
--export([make_key/2]).
+-export([make_key/2, filter/1]).
 -export([match/2, matches/3, get_id/1, get_topic/1]).
--export_type([key/1, word/0, nextf/0, opts/0]).
+-export_type([key/1, word/0, words/0, nextf/0, opts/0]).
 
 -define(END, '$end_of_table').
 
 -type word() :: binary() | '+' | '#'.
+-type words() :: [word()].
 -type base_key() :: {binary() | [word()], {}}.
 -type key(ID) :: {binary() | [word()], {ID}}.
 -type nextf() :: fun((key(_) | base_key()) -> ?END | key(_)).
 -type opts() :: [unique | return_first].
 
 %% @doc Make a search-key for the given topic.
--spec make_key(emqx_types:topic(), ID) -> key(ID).
+-spec make_key(emqx_types:topic() | words(), ID) -> key(ID).
 make_key(Topic, ID) when is_binary(Topic) ->
-    Words = filter_words(Topic),
-    case emqx_topic:wildcard(Words) of
-        true ->
+    case filter(Topic) of
+        Words when is_list(Words) ->
             %% it's a wildcard
             {Words, {ID}};
         false ->
@@ -123,7 +123,15 @@ make_key(Topic, ID) when is_binary(Topic) ->
             %% because they can be found with direct lookups.
             %% it is also more compact in memory.
             {Topic, {ID}}
-    end.
+    end;
+make_key(Words, ID) when is_list(Words) ->
+    {Words, {ID}}.
+
+%% @doc Parse a topic filter into a list of words. Returns `false` if it's not a filter.
+-spec filter(emqx_types:topic()) -> words() | false.
+filter(Topic) ->
+    Words = filter_words(Topic),
+    emqx_topic:wildcard(Words) andalso Words.
 
 %% @doc Extract record ID from the match.
 -spec get_id(key(ID)) -> ID.
@@ -325,6 +333,7 @@ filter_words(Topic) when is_binary(Topic) ->
     % `match_filter/3` expects.
     [word(W, filter) || W <- emqx_topic:tokens(Topic)].
 
+-spec topic_words(emqx_types:topic()) -> [binary()].
 topic_words(Topic) when is_binary(Topic) ->
     [word(W, topic) || W <- emqx_topic:tokens(Topic)].
 
diff --git a/apps/emqx/test/emqx_cth_cluster.erl b/apps/emqx/test/emqx_cth_cluster.erl
index 9ad8dc0f4..e1dc9d977 100644
--- a/apps/emqx/test/emqx_cth_cluster.erl
+++ b/apps/emqx/test/emqx_cth_cluster.erl
@@ -19,9 +19,13 @@
 -export([start/2]).
 -export([stop/1]).
 
+-export([start_bare_node/2]).
+
 -export([share_load_module/2]).
 -export([node_name/1]).
 
+-export([node_name/1]).
+
 -define(APPS_CLUSTERING, [gen_rpc, mria, ekka]).
 
 -define(TIMEOUT_NODE_START_MS, 15000).
@@ -259,9 +263,6 @@ allocate_listener_ports(Types, Spec) ->
 
 start_node_init(Spec = #{name := Node}) ->
     Node = start_bare_node(Node, Spec),
-    pong = net_adm:ping(Node),
-    % Preserve node spec right on the remote node
-    ok = set_node_opts(Node, Spec),
     % Make it possible to call `ct:pal` and friends (if running under rebar3)
     _ = share_load_module(Node, cthr),
     % Enable snabbkaffe trace forwarding
@@ -366,7 +367,8 @@ listener_port(BasePort, wss) ->
 
 %%
 
-start_bare_node(Name, #{driver := ct_slave}) ->
+-spec start_bare_node(atom(), map()) -> node().
+start_bare_node(Name, Spec = #{driver := ct_slave}) ->
     {ok, Node} = ct_slave:start(
         node_name(Name),
         [
@@ -378,9 +380,15 @@ start_bare_node(Name, #{driver := ct_slave}) ->
             {env, []}
         ]
     ),
-    Node;
-start_bare_node(Name, #{driver := slave}) ->
+    init_bare_node(Node, Spec);
+start_bare_node(Name, Spec = #{driver := slave}) ->
     {ok, Node} = slave:start_link(host(), Name, ebin_path()),
+    init_bare_node(Node, Spec).
+
+init_bare_node(Node, Spec) ->
+    pong = net_adm:ping(Node),
+    % Preserve node spec right on the remote node
+    ok = set_node_opts(Node, Spec),
     Node.
 
 erl_flags() ->
@@ -403,6 +411,7 @@ share_load_module(Node, Module) ->
             error
     end.
 
+-spec node_name(atom()) -> node().
 node_name(Name) ->
     case string:tokens(atom_to_list(Name), "@") of
         [_Name, _Host] ->
diff --git a/apps/emqx/test/emqx_cth_suite.erl b/apps/emqx/test/emqx_cth_suite.erl
index 307e8d98b..e5668b77a 100644
--- a/apps/emqx/test/emqx_cth_suite.erl
+++ b/apps/emqx/test/emqx_cth_suite.erl
@@ -261,7 +261,7 @@ merge_envs(false, E2) ->
 merge_envs(_E, false) ->
     [];
 merge_envs(E1, E2) ->
-    E1 ++ E2.
+    lists:foldl(fun({K, _} = Opt, EAcc) -> lists:keystore(K, 1, EAcc, Opt) end, E1, E2).
 
 merge_config(false, C2) ->
     C2;
diff --git a/apps/emqx/test/emqx_quic_multistreams_SUITE.erl b/apps/emqx/test/emqx_quic_multistreams_SUITE.erl
index 1b45cb669..267782ff9 100644
--- a/apps/emqx/test/emqx_quic_multistreams_SUITE.erl
+++ b/apps/emqx/test/emqx_quic_multistreams_SUITE.erl
@@ -1094,7 +1094,7 @@ t_multi_streams_unsub(Config) ->
     ?retry(
         _Sleep2 = 100,
         _Attempts2 = 50,
-        false = emqx_router:has_routes(Topic)
+        [] = emqx_router:lookup_routes(Topic)
     ),
 
     case emqtt:publish_via(C, PubVia, Topic, #{}, <<6, 7, 8, 9>>, [{qos, PubQos}]) of
diff --git a/apps/emqx/test/emqx_router_SUITE.erl b/apps/emqx/test/emqx_router_SUITE.erl
index 067f11634..1128112ff 100644
--- a/apps/emqx/test/emqx_router_SUITE.erl
+++ b/apps/emqx/test/emqx_router_SUITE.erl
@@ -26,24 +26,37 @@
 
 -define(R, emqx_router).
 
-all() -> emqx_common_test_helpers:all(?MODULE).
-
-init_per_suite(Config) ->
-    PrevBootModules = application:get_env(emqx, boot_modules),
-    emqx_common_test_helpers:boot_modules([router]),
-    emqx_common_test_helpers:start_apps([]),
+all() ->
     [
-        {prev_boot_modules, PrevBootModules}
-        | Config
+        {group, routing_schema_v1},
+        {group, routing_schema_v2}
     ].
 
-end_per_suite(Config) ->
-    PrevBootModules = ?config(prev_boot_modules, Config),
-    case PrevBootModules of
-        undefined -> ok;
-        {ok, Mods} -> emqx_common_test_helpers:boot_modules(Mods)
-    end,
-    emqx_common_test_helpers:stop_apps([]).
+groups() ->
+    TCs = emqx_common_test_helpers:all(?MODULE),
+    [
+        {routing_schema_v1, [], TCs},
+        {routing_schema_v2, [], TCs}
+    ].
+
+init_per_group(GroupName, Config) ->
+    WorkDir = filename:join([?config(priv_dir, Config), ?MODULE, GroupName]),
+    AppSpecs = [
+        {emqx, #{
+            config => mk_config(GroupName),
+            override_env => [{boot_modules, [router]}]
+        }}
+    ],
+    Apps = emqx_cth_suite:start(AppSpecs, #{work_dir => WorkDir}),
+    [{group_apps, Apps}, {group_name, GroupName} | Config].
+
+end_per_group(_GroupName, Config) ->
+    ok = emqx_cth_suite:stop(?config(group_apps, Config)).
+
+mk_config(routing_schema_v1) ->
+    "broker.routing.storage_schema = v1";
+mk_config(routing_schema_v2) ->
+    "broker.routing.storage_schema = v2".
 
 init_per_testcase(_TestCase, Config) ->
     clear_tables(),
@@ -52,23 +65,16 @@ init_per_testcase(_TestCase, Config) ->
 end_per_testcase(_TestCase, _Config) ->
     clear_tables().
 
-% t_add_route(_) ->
-%     error('TODO').
-
-% t_do_add_route(_) ->
-%     error('TODO').
-
 % t_lookup_routes(_) ->
 %     error('TODO').
 
-% t_delete_route(_) ->
-%     error('TODO').
-
-% t_do_delete_route(_) ->
-%     error('TODO').
-
-% t_topics(_) ->
-%     error('TODO').
+t_verify_type(Config) ->
+    case ?config(group_name, Config) of
+        routing_schema_v1 ->
+            ?assertEqual(v1, ?R:get_schema_vsn());
+        routing_schema_v2 ->
+            ?assertEqual(v2, ?R:get_schema_vsn())
+    end.
 
 t_add_delete(_) ->
     ?R:add_route(<<"a/b/c">>),
@@ -79,6 +85,55 @@ t_add_delete(_) ->
     ?R:delete_route(<<"a/+/b">>, node()),
     ?assertEqual([], ?R:topics()).
 
+t_add_delete_incremental(_) ->
+    ?R:add_route(<<"a/b/c">>),
+    ?R:add_route(<<"a/+/c">>, node()),
+    ?R:add_route(<<"a/+/+">>, node()),
+    ?R:add_route(<<"a/b/#">>, node()),
+    ?R:add_route(<<"#">>, node()),
+    ?assertEqual(
+        [
+            #route{topic = <<"#">>, dest = node()},
+            #route{topic = <<"a/+/+">>, dest = node()},
+            #route{topic = <<"a/+/c">>, dest = node()},
+            #route{topic = <<"a/b/#">>, dest = node()},
+            #route{topic = <<"a/b/c">>, dest = node()}
+        ],
+        lists:sort(?R:match_routes(<<"a/b/c">>))
+    ),
+    ?R:delete_route(<<"a/+/c">>, node()),
+    ?assertEqual(
+        [
+            #route{topic = <<"#">>, dest = node()},
+            #route{topic = <<"a/+/+">>, dest = node()},
+            #route{topic = <<"a/b/#">>, dest = node()},
+            #route{topic = <<"a/b/c">>, dest = node()}
+        ],
+        lists:sort(?R:match_routes(<<"a/b/c">>))
+    ),
+    ?R:delete_route(<<"a/+/+">>, node()),
+    ?assertEqual(
+        [
+            #route{topic = <<"#">>, dest = node()},
+            #route{topic = <<"a/b/#">>, dest = node()},
+            #route{topic = <<"a/b/c">>, dest = node()}
+        ],
+        lists:sort(?R:match_routes(<<"a/b/c">>))
+    ),
+    ?R:delete_route(<<"a/b/#">>, node()),
+    ?assertEqual(
+        [
+            #route{topic = <<"#">>, dest = node()},
+            #route{topic = <<"a/b/c">>, dest = node()}
+        ],
+        lists:sort(?R:match_routes(<<"a/b/c">>))
+    ),
+    ?R:delete_route(<<"a/b/c">>, node()),
+    ?assertEqual(
+        [#route{topic = <<"#">>, dest = node()}],
+        lists:sort(?R:match_routes(<<"a/b/c">>))
+    ).
+
 t_do_add_delete(_) ->
     ?R:do_add_route(<<"a/b/c">>),
     ?R:do_add_route(<<"a/b/c">>, node()),
@@ -114,9 +169,9 @@ t_print_routes(_) ->
     ?R:add_route(<<"+/+">>),
     ?R:print_routes(<<"a/b">>).
 
-t_has_routes(_) ->
+t_has_route(_) ->
     ?R:add_route(<<"devices/+/messages">>, node()),
-    ?assert(?R:has_routes(<<"devices/+/messages">>)),
+    ?assert(?R:has_route(<<"devices/+/messages">>, node())),
     ?R:delete_route(<<"devices/+/messages">>).
 
 t_unexpected(_) ->
@@ -128,5 +183,5 @@ t_unexpected(_) ->
 clear_tables() ->
     lists:foreach(
         fun mnesia:clear_table/1,
-        [?ROUTE_TAB, ?TRIE, emqx_trie_node]
+        [?ROUTE_TAB, ?ROUTE_TAB_FILTERS, ?TRIE]
     ).
diff --git a/apps/emqx/test/emqx_router_helper_SUITE.erl b/apps/emqx/test/emqx_router_helper_SUITE.erl
index c0796288e..889c8293c 100644
--- a/apps/emqx/test/emqx_router_helper_SUITE.erl
+++ b/apps/emqx/test/emqx_router_helper_SUITE.erl
@@ -26,55 +26,45 @@
 
 -define(ROUTER_HELPER, emqx_router_helper).
 
-all() -> emqx_common_test_helpers:all(?MODULE).
+all() ->
+    [
+        {group, routing_schema_v1},
+        {group, routing_schema_v2}
+    ].
 
-init_per_suite(Config) ->
-    DistPid =
-        case net_kernel:nodename() of
-            ignored ->
-                %% calling `net_kernel:start' without `epmd'
-                %% running will result in a failure.
-                emqx_common_test_helpers:start_epmd(),
-                {ok, Pid} = net_kernel:start(['test@127.0.0.1', longnames]),
-                Pid;
-            _ ->
-                undefined
-        end,
-    emqx_common_test_helpers:start_apps([]),
-    [{dist_pid, DistPid} | Config].
+groups() ->
+    TCs = emqx_common_test_helpers:all(?MODULE),
+    [
+        {routing_schema_v1, [], TCs},
+        {routing_schema_v2, [], TCs}
+    ].
 
-end_per_suite(Config) ->
-    DistPid = ?config(dist_pid, Config),
-    case DistPid of
-        Pid when is_pid(Pid) ->
-            net_kernel:stop();
-        _ ->
-            ok
-    end,
-    emqx_common_test_helpers:stop_apps([]).
+init_per_group(GroupName, Config) ->
+    WorkDir = filename:join([?config(priv_dir, Config), ?MODULE, GroupName]),
+    AppSpecs = [{emqx, mk_config(GroupName)}],
+    Apps = emqx_cth_suite:start(AppSpecs, #{work_dir => WorkDir}),
+    [{group_name, GroupName}, {group_apps, Apps} | Config].
+
+end_per_group(_GroupName, Config) ->
+    ok = emqx_cth_suite:stop(?config(group_apps, Config)).
+
+mk_config(routing_schema_v1) ->
+    #{
+        config => "broker.routing.storage_schema = v1",
+        override_env => [{boot_modules, [router]}]
+    };
+mk_config(routing_schema_v2) ->
+    #{
+        config => "broker.routing.storage_schema = v2",
+        override_env => [{boot_modules, [router]}]
+    }.
 
-init_per_testcase(TestCase, Config) when
-    TestCase =:= t_cleanup_membership_mnesia_down;
-    TestCase =:= t_cleanup_membership_node_down;
-    TestCase =:= t_cleanup_monitor_node_down
-->
-    ok = snabbkaffe:start_trace(),
-    Slave = emqx_common_test_helpers:start_slave(some_node, []),
-    [{slave, Slave} | Config];
 init_per_testcase(_TestCase, Config) ->
+    ok = snabbkaffe:start_trace(),
     Config.
 
-end_per_testcase(TestCase, Config) when
-    TestCase =:= t_cleanup_membership_mnesia_down;
-    TestCase =:= t_cleanup_membership_node_down;
-    TestCase =:= t_cleanup_monitor_node_down
-->
-    Slave = ?config(slave, Config),
-    emqx_common_test_helpers:stop_slave(Slave),
-    mria:clear_table(?ROUTE_TAB),
-    snabbkaffe:stop(),
-    ok;
 end_per_testcase(_TestCase, _Config) ->
+    ok = snabbkaffe:stop(),
     ok.
 
 t_monitor(_) ->
@@ -89,8 +79,8 @@ t_mnesia(_) ->
     ?ROUTER_HELPER ! {membership, {mnesia, down, node()}},
     ct:sleep(200).
 
-t_cleanup_membership_mnesia_down(Config) ->
-    Slave = ?config(slave, Config),
+t_cleanup_membership_mnesia_down(_Config) ->
+    Slave = emqx_cth_cluster:node_name(?FUNCTION_NAME),
     emqx_router:add_route(<<"a/b/c">>, Slave),
     emqx_router:add_route(<<"d/e/f">>, node()),
     ?assertMatch([_, _], emqx_router:topics()),
@@ -101,8 +91,8 @@ t_cleanup_membership_mnesia_down(Config) ->
     ),
     ?assertEqual([<<"d/e/f">>], emqx_router:topics()).
 
-t_cleanup_membership_node_down(Config) ->
-    Slave = ?config(slave, Config),
+t_cleanup_membership_node_down(_Config) ->
+    Slave = emqx_cth_cluster:node_name(?FUNCTION_NAME),
     emqx_router:add_route(<<"a/b/c">>, Slave),
     emqx_router:add_route(<<"d/e/f">>, node()),
     ?assertMatch([_, _], emqx_router:topics()),
@@ -113,13 +103,13 @@ t_cleanup_membership_node_down(Config) ->
     ),
     ?assertEqual([<<"d/e/f">>], emqx_router:topics()).
 
-t_cleanup_monitor_node_down(Config) ->
-    Slave = ?config(slave, Config),
+t_cleanup_monitor_node_down(_Config) ->
+    Slave = emqx_cth_cluster:start_bare_node(?FUNCTION_NAME, #{driver => ct_slave}),
     emqx_router:add_route(<<"a/b/c">>, Slave),
     emqx_router:add_route(<<"d/e/f">>, node()),
     ?assertMatch([_, _], emqx_router:topics()),
     ?wait_async_action(
-        emqx_common_test_helpers:stop_slave(Slave),
+        emqx_cth_cluster:stop([Slave]),
         #{?snk_kind := emqx_router_helper_cleanup_done, node := Slave},
         1_000
     ),
diff --git a/apps/emqx/test/emqx_routing_SUITE.erl b/apps/emqx/test/emqx_routing_SUITE.erl
new file mode 100644
index 000000000..6966ac56a
--- /dev/null
+++ b/apps/emqx/test/emqx_routing_SUITE.erl
@@ -0,0 +1,258 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2017-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_routing_SUITE).
+
+-compile(export_all).
+-compile(nowarn_export_all).
+
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
+-include_lib("emqx/include/asserts.hrl").
+
+all() ->
+    [
+        {group, routing_schema_v1},
+        {group, routing_schema_v2},
+        t_routing_schema_switch_v1,
+        t_routing_schema_switch_v2
+    ].
+
+groups() ->
+    TCs = [
+        t_cluster_routing
+    ],
+    [
+        {routing_schema_v1, [], TCs},
+        {routing_schema_v2, [], TCs}
+    ].
+
+init_per_group(GroupName, Config) ->
+    WorkDir = filename:join([?config(priv_dir, Config), ?MODULE, GroupName]),
+    NodeSpecs = [
+        {emqx_routing_SUITE1, #{apps => [mk_emqx_appspec(GroupName, 1)], role => core}},
+        {emqx_routing_SUITE2, #{apps => [mk_emqx_appspec(GroupName, 2)], role => core}},
+        {emqx_routing_SUITE3, #{apps => [mk_emqx_appspec(GroupName, 3)], role => replicant}}
+    ],
+    Nodes = emqx_cth_cluster:start(NodeSpecs, #{work_dir => WorkDir}),
+    [{cluster, Nodes} | Config].
+
+end_per_group(_GroupName, Config) ->
+    emqx_cth_cluster:stop(?config(cluster, Config)).
+
+init_per_testcase(TC, Config) ->
+    WorkDir = filename:join([?config(priv_dir, Config), ?MODULE, TC]),
+    [{work_dir, WorkDir} | Config].
+
+end_per_testcase(_TC, _Config) ->
+    ok.
+
+mk_emqx_appspec(GroupName, N) ->
+    {emqx, #{
+        config => mk_config(GroupName, N),
+        after_start => fun() ->
+            % NOTE
+            % This one is actually defined on `emqx_conf_schema` level, but used
+            % in `emqx_broker`. Thus we have to resort to this ugly hack.
+            emqx_config:force_put([rpc, mode], async)
+        end
+    }}.
+
+mk_genrpc_appspec() ->
+    {gen_rpc, #{
+        override_env => [{port_discovery, stateless}]
+    }}.
+
+mk_config(GroupName, N) ->
+    #{
+        broker => mk_config_broker(GroupName),
+        listeners => mk_config_listeners(N)
+    }.
+
+mk_config_broker(Vsn) when Vsn == routing_schema_v1; Vsn == v1 ->
+    #{routing => #{storage_schema => v1}};
+mk_config_broker(Vsn) when Vsn == routing_schema_v2; Vsn == v2 ->
+    #{routing => #{storage_schema => v2}}.
+
+mk_config_listeners(N) ->
+    Port = 1883 + N,
+    #{
+        tcp => #{default => #{bind => "127.0.0.1:" ++ integer_to_list(Port)}},
+        ssl => #{default => #{enable => false}},
+        ws => #{default => #{enable => false}},
+        wss => #{default => #{enable => false}}
+    }.
+
+%%
+
+t_cluster_routing(Config) ->
+    Cluster = ?config(cluster, Config),
+    Clients = [C1, C2, C3] = [start_client(N) || N <- Cluster],
+    Commands = [
+        {fun publish/3, [C1, <<"a/b/c">>, <<"wontsee">>]},
+        {fun publish/3, [C2, <<"a/b/d">>, <<"wontsee">>]},
+        {fun subscribe/2, [C3, <<"a/+/c/#">>]},
+        {fun publish/3, [C1, <<"a/b/c">>, <<"01">>]},
+        {fun publish/3, [C2, <<"a/b/d">>, <<"wontsee">>]},
+        {fun subscribe/2, [C1, <<"a/b/c">>]},
+        {fun subscribe/2, [C2, <<"a/b/+">>]},
+        {fun publish/3, [C3, <<"a/b/c">>, <<"02">>]},
+        {fun publish/3, [C2, <<"a/b/d">>, <<"03">>]},
+        {fun publish/3, [C2, <<"a/b/c/d">>, <<"04">>]},
+        {fun subscribe/2, [C3, <<"a/b/d">>]},
+        {fun publish/3, [C1, <<"a/b/d">>, <<"05">>]},
+        {fun unsubscribe/2, [C3, <<"a/+/c/#">>]},
+        {fun publish/3, [C1, <<"a/b/c">>, <<"06">>]},
+        {fun publish/3, [C2, <<"a/b/d">>, <<"07">>]},
+        {fun publish/3, [C2, <<"a/b/c/d">>, <<"08">>]},
+        {fun unsubscribe/2, [C2, <<"a/b/+">>]},
+        {fun publish/3, [C1, <<"a/b/c">>, <<"09">>]},
+        {fun publish/3, [C2, <<"a/b/d">>, <<"10">>]},
+        {fun publish/3, [C2, <<"a/b/c/d">>, <<"11">>]},
+        {fun unsubscribe/2, [C3, <<"a/b/d">>]},
+        {fun unsubscribe/2, [C1, <<"a/b/c">>]},
+        {fun publish/3, [C1, <<"a/b/c">>, <<"wontsee">>]},
+        {fun publish/3, [C2, <<"a/b/d">>, <<"wontsee">>]}
+    ],
+    ok = lists:foreach(fun({F, Args}) -> erlang:apply(F, Args) end, Commands),
+    _ = [emqtt:stop(C) || C <- Clients],
+    Deliveries = ?drainMailbox(),
+    ?assertMatch(
+        [
+            {pub, C1, #{topic := <<"a/b/c">>, payload := <<"02">>}},
+            {pub, C1, #{topic := <<"a/b/c">>, payload := <<"06">>}},
+            {pub, C1, #{topic := <<"a/b/c">>, payload := <<"09">>}},
+            {pub, C2, #{topic := <<"a/b/c">>, payload := <<"02">>}},
+            {pub, C2, #{topic := <<"a/b/d">>, payload := <<"03">>}},
+            {pub, C2, #{topic := <<"a/b/d">>, payload := <<"05">>}},
+            {pub, C2, #{topic := <<"a/b/c">>, payload := <<"06">>}},
+            {pub, C2, #{topic := <<"a/b/d">>, payload := <<"07">>}},
+            {pub, C3, #{topic := <<"a/b/c">>, payload := <<"01">>}},
+            {pub, C3, #{topic := <<"a/b/c">>, payload := <<"02">>}},
+            {pub, C3, #{topic := <<"a/b/c/d">>, payload := <<"04">>}},
+            {pub, C3, #{topic := <<"a/b/d">>, payload := <<"05">>}},
+            {pub, C3, #{topic := <<"a/b/d">>, payload := <<"07">>}},
+            {pub, C3, #{topic := <<"a/b/d">>, payload := <<"10">>}}
+        ],
+        lists:sort(
+            fun({pub, CL, #{payload := PL}}, {pub, CR, #{payload := PR}}) ->
+                {CL, PL} < {CR, PR}
+            end,
+            Deliveries
+        )
+    ).
+
+start_client(Node) ->
+    Self = self(),
+    {ok, C} = emqtt:start_link(#{
+        port => get_mqtt_tcp_port(Node),
+        msg_handler => #{
+            publish => fun(Msg) -> Self ! {pub, self(), Msg} end
+        }
+    }),
+    {ok, _Props} = emqtt:connect(C),
+    C.
+
+publish(C, Topic, Payload) ->
+    {ok, #{reason_code := 0}} = emqtt:publish(C, Topic, Payload, 1).
+
+subscribe(C, Topic) ->
+    % NOTE: sleeping here as lazy way to wait for subscribe to replicate
+    {ok, _Props, [0]} = emqtt:subscribe(C, Topic),
+    ok = timer:sleep(200).
+
+unsubscribe(C, Topic) ->
+    % NOTE: sleeping here as lazy way to wait for unsubscribe to replicate
+    {ok, _Props, undefined} = emqtt:unsubscribe(C, Topic),
+    ok = timer:sleep(200).
+
+%%
+
+t_routing_schema_switch_v1(Config) ->
+    t_routing_schema_switch(_From = v2, _To = v1, Config).
+
+t_routing_schema_switch_v2(Config) ->
+    t_routing_schema_switch(_From = v1, _To = v2, Config).
+
+t_routing_schema_switch(VFrom, VTo, Config) ->
+    % Start first node with routing schema VTo (e.g. v1)
+    WorkDir = ?config(work_dir, Config),
+    [Node1] = emqx_cth_cluster:start(
+        [
+            {routing_schema_switch1, #{
+                apps => [mk_genrpc_appspec(), mk_emqx_appspec(VTo, 1)]
+            }}
+        ],
+        #{work_dir => WorkDir}
+    ),
+    % Ensure there's at least 1 route on Node1
+    C1 = start_client(Node1),
+    ok = subscribe(C1, <<"a/+/c">>),
+    ok = subscribe(C1, <<"d/e/f/#">>),
+    % Start rest of nodes with routing schema VFrom (e.g. v2)
+    [Node2, Node3] = emqx_cth_cluster:start(
+        [
+            {routing_schema_switch2, #{
+                apps => [mk_genrpc_appspec(), mk_emqx_appspec(VFrom, 2)],
+                base_port => 20000,
+                join_to => Node1
+            }},
+            {routing_schema_switch3, #{
+                apps => [mk_genrpc_appspec(), mk_emqx_appspec(VFrom, 3)],
+                base_port => 20100,
+                join_to => Node1
+            }}
+        ],
+        #{work_dir => WorkDir}
+    ),
+    % Verify that new nodes switched to schema v1/v2 in presence of v1/v2 routes respectively
+    Nodes = [Node1, Node2, Node3],
+    ?assertEqual(
+        [{ok, VTo}, {ok, VTo}, {ok, VTo}],
+        erpc:multicall(Nodes, emqx_router, get_schema_vsn, [])
+    ),
+    % Wait for all nodes to agree on cluster state
+    ?retry(
+        500,
+        10,
+        ?assertMatch(
+            [{ok, [Node1, Node2, Node3]}],
+            lists:usort(erpc:multicall(Nodes, emqx, running_nodes, []))
+        )
+    ),
+    % Verify that routing works as expected
+    C2 = start_client(Node2),
+    ok = subscribe(C2, <<"a/+/d">>),
+    C3 = start_client(Node3),
+    ok = subscribe(C3, <<"d/e/f/#">>),
+    {ok, _} = publish(C1, <<"a/b/d">>, <<"hey-newbies">>),
+    {ok, _} = publish(C2, <<"a/b/c">>, <<"hi">>),
+    {ok, _} = publish(C3, <<"d/e/f/42">>, <<"hello">>),
+    ?assertReceive({pub, C2, #{topic := <<"a/b/d">>, payload := <<"hey-newbies">>}}),
+    ?assertReceive({pub, C1, #{topic := <<"a/b/c">>, payload := <<"hi">>}}),
+    ?assertReceive({pub, C1, #{topic := <<"d/e/f/42">>, payload := <<"hello">>}}),
+    ?assertReceive({pub, C3, #{topic := <<"d/e/f/42">>, payload := <<"hello">>}}),
+    ?assertNotReceive(_),
+    ok = emqtt:stop(C1),
+    ok = emqtt:stop(C2),
+    ok = emqtt:stop(C3),
+    ok = emqx_cth_cluster:stop(Nodes).
+
+%%
+
+get_mqtt_tcp_port(Node) ->
+    {_, Port} = erpc:call(Node, emqx_config, get, [[listeners, tcp, default, bind]]),
+    Port.
diff --git a/apps/emqx/test/emqx_shared_sub_SUITE.erl b/apps/emqx/test/emqx_shared_sub_SUITE.erl
index 6439981f6..7a7729878 100644
--- a/apps/emqx/test/emqx_shared_sub_SUITE.erl
+++ b/apps/emqx/test/emqx_shared_sub_SUITE.erl
@@ -1054,7 +1054,7 @@ t_queue_subscription(Config) when is_list(Config) ->
         begin
             ct:pal("routes: ~p", [ets:tab2list(emqx_route)]),
             %% FIXME: should ensure we have 2 subscriptions
-            true = emqx_router:has_routes(Topic)
+            [_] = emqx_router:lookup_routes(Topic)
         end
     ),
 
@@ -1081,7 +1081,7 @@ t_queue_subscription(Config) when is_list(Config) ->
     %%     _Attempts0 = 50,
     %%    begin
     %%     ct:pal("routes: ~p", [ets:tab2list(emqx_route)]),
-    %%     false = emqx_router:has_routes(Topic)
+    %%     [] = emqx_router:lookup_routes(Topic)
     %%    end
     %%   ),
     ct:sleep(500),
diff --git a/apps/emqx/test/emqx_topic_index_SUITE.erl b/apps/emqx/test/emqx_topic_index_SUITE.erl
index 08056a16f..9df9743f1 100644
--- a/apps/emqx/test/emqx_topic_index_SUITE.erl
+++ b/apps/emqx/test/emqx_topic_index_SUITE.erl
@@ -57,6 +57,17 @@ t_insert(Config) ->
     ?assertEqual(<<"sensor/#">>, topic(match(M, <<"sensor">>, Tab))),
     ?assertEqual(t_insert_3, id(match(M, <<"sensor">>, Tab))).
 
+t_insert_filter(Config) ->
+    M = get_module(Config),
+    Tab = M:new(),
+    Topic = <<"sensor/+/metric//#">>,
+    true = M:insert(Topic, 1, <<>>, Tab),
+    true = M:insert(emqx_trie_search:filter(Topic), 2, <<>>, Tab),
+    ?assertEqual(
+        [Topic, Topic],
+        [topic(X) || X <- matches(M, <<"sensor/1/metric//2">>, Tab)]
+    ).
+
 t_match(Config) ->
     M = get_module(Config),
     Tab = M:new(),
diff --git a/apps/emqx/test/emqx_trie_search_tests.erl b/apps/emqx/test/emqx_trie_search_tests.erl
index 75994131d..d78347de6 100644
--- a/apps/emqx/test/emqx_trie_search_tests.erl
+++ b/apps/emqx/test/emqx_trie_search_tests.erl
@@ -18,15 +18,30 @@
 
 -include_lib("eunit/include/eunit.hrl").
 
-topic_validation_test() ->
+-import(emqx_trie_search, [filter/1]).
+
+filter_test_() ->
+    [
+        ?_assertEqual(
+            [<<"sensor">>, '+', <<"metric">>, <<>>, '#'],
+            filter(<<"sensor/+/metric//#">>)
+        ),
+        ?_assertEqual(
+            false,
+            filter(<<"sensor/1/metric//42">>)
+        )
+    ].
+
+topic_validation_test_() ->
     NextF = fun(_) -> '$end_of_table' end,
     Call = fun(Topic) ->
         emqx_trie_search:match(Topic, NextF)
     end,
-    ?assertError(badarg, Call(<<"+">>)),
-    ?assertError(badarg, Call(<<"#">>)),
-    ?assertError(badarg, Call(<<"a/+/b">>)),
-    ?assertError(badarg, Call(<<"a/b/#">>)),
-    ?assertEqual(false, Call(<<"a/b/b+">>)),
-    ?assertEqual(false, Call(<<"a/b/c#">>)),
-    ok.
+    [
+        ?_assertError(badarg, Call(<<"+">>)),
+        ?_assertError(badarg, Call(<<"#">>)),
+        ?_assertError(badarg, Call(<<"a/+/b">>)),
+        ?_assertError(badarg, Call(<<"a/b/#">>)),
+        ?_assertEqual(false, Call(<<"a/b/b+">>)),
+        ?_assertEqual(false, Call(<<"a/b/c#">>))
+    ].
diff --git a/apps/emqx_bridge_kafka/test/emqx_bridge_kafka_impl_consumer_SUITE.erl b/apps/emqx_bridge_kafka/test/emqx_bridge_kafka_impl_consumer_SUITE.erl
index 2f5840f41..1691fa6a4 100644
--- a/apps/emqx_bridge_kafka/test/emqx_bridge_kafka_impl_consumer_SUITE.erl
+++ b/apps/emqx_bridge_kafka/test/emqx_bridge_kafka_impl_consumer_SUITE.erl
@@ -1929,7 +1929,7 @@ t_node_joins_existing_cluster(Config) ->
             ?retry(
                 _Sleep2 = 100,
                 _Attempts2 = 50,
-                true = erpc:call(N2, emqx_router, has_routes, [MQTTTopic])
+                [] =/= erpc:call(N2, emqx_router, lookup_routes, [MQTTTopic])
             ),
             {ok, SRef1} =
                 snabbkaffe:subscribe(
diff --git a/apps/emqx_eviction_agent/test/emqx_eviction_agent_test_helpers.erl b/apps/emqx_eviction_agent/test/emqx_eviction_agent_test_helpers.erl
index 130a2628a..860436f67 100644
--- a/apps/emqx_eviction_agent/test/emqx_eviction_agent_test_helpers.erl
+++ b/apps/emqx_eviction_agent/test/emqx_eviction_agent_test_helpers.erl
@@ -85,7 +85,7 @@ start_cluster(NamesWithPorts, Apps, Env) ->
         NamesWithPorts
     ),
     Opts0 = [
-        {env, [{emqx, boot_modules, [broker, listeners]}] ++ Env},
+        {env, Env},
         {apps, Apps},
         {conf,
             [{[listeners, Proto, default, enable], false} || Proto <- [ssl, ws, wss]] ++
diff --git a/apps/emqx_management/src/emqx_mgmt_cli.erl b/apps/emqx_management/src/emqx_mgmt_cli.erl
index aeed5b922..121e5b1a8 100644
--- a/apps/emqx_management/src/emqx_mgmt_cli.erl
+++ b/apps/emqx_management/src/emqx_mgmt_cli.erl
@@ -22,9 +22,6 @@
 -include_lib("emqx/include/emqx_mqtt.hrl").
 -include_lib("emqx/include/logger.hrl").
 
--include("emqx_mgmt.hrl").
-
--define(PRINT_CMD(Cmd, Descr), io:format("~-48s# ~ts~n", [Cmd, Descr])).
 -define(DATA_BACKUP_OPTS, #{print_fun => fun emqx_ctl:print/2}).
 
 -export([load/0]).
@@ -49,20 +46,6 @@
     data/1
 ]).
 
--define(PROC_INFOKEYS, [
-    status,
-    memory,
-    message_queue_len,
-    total_heap_size,
-    heap_size,
-    stack_size,
-    reductions
-]).
-
--define(MAX_LIMIT, 10000).
-
--define(APP, emqx).
-
 -spec load() -> ok.
 load() ->
     Cmds = [Fun || {Fun, _} <- ?MODULE:module_info(exports), is_cmd(Fun)],
@@ -197,9 +180,12 @@ if_client(ClientId, Fun) ->
 %% @doc Topics Command
 
 topics(["list"]) ->
-    dump(?ROUTE_TAB, emqx_topic);
+    emqx_router:foldr_routes(
+        fun(Route, Acc) -> [print({emqx_topic, Route}) | Acc] end,
+        []
+    );
 topics(["show", Topic]) ->
-    Routes = ets:lookup(?ROUTE_TAB, bin(Topic)),
+    Routes = emqx_router:lookup_routes(Topic),
     [print({emqx_topic, Route}) || Route <- Routes];
 topics(_) ->
     emqx_ctl:usage([
diff --git a/apps/emqx_rule_engine/src/emqx_rule_engine.erl b/apps/emqx_rule_engine/src/emqx_rule_engine.erl
index 41d1ed433..01b9b76d0 100644
--- a/apps/emqx_rule_engine/src/emqx_rule_engine.erl
+++ b/apps/emqx_rule_engine/src/emqx_rule_engine.erl
@@ -225,8 +225,9 @@ get_rules_ordered_by_ts() ->
 -spec get_rules_for_topic(Topic :: binary()) -> [rule()].
 get_rules_for_topic(Topic) ->
     [
-        emqx_topic_index:get_record(M, ?RULE_TOPIC_INDEX)
-     || M <- emqx_topic_index:matches(Topic, ?RULE_TOPIC_INDEX, [unique])
+        Rule
+     || M <- emqx_topic_index:matches(Topic, ?RULE_TOPIC_INDEX, [unique]),
+        Rule <- lookup_rule(emqx_topic_index:get_id(M))
     ].
 
 -spec get_rules_with_same_event(Topic :: binary()) -> [rule()].
@@ -284,11 +285,14 @@ is_of_event_name(EventName, Topic) ->
 
 -spec get_rule(Id :: rule_id()) -> {ok, rule()} | not_found.
 get_rule(Id) ->
-    case ets:lookup(?RULE_TAB, Id) of
-        [{Id, Rule}] -> {ok, Rule#{id => Id}};
+    case lookup_rule(Id) of
+        [Rule] -> {ok, Rule};
         [] -> not_found
     end.
 
+lookup_rule(Id) ->
+    [Rule || {_Id, Rule} <- ets:lookup(?RULE_TAB, Id)].
+
 load_hooks_for_rule(#{from := Topics}) ->
     lists:foreach(fun emqx_rule_events:load/1, Topics).
 
@@ -483,7 +487,7 @@ with_parsed_rule(Params = #{id := RuleId, sql := Sql, actions := Actions}, Creat
 do_insert_rule(#{id := Id} = Rule) ->
     ok = load_hooks_for_rule(Rule),
     ok = maybe_add_metrics_for_rule(Id),
-    true = ets:insert(?RULE_TAB, {Id, maps:remove(id, Rule)}),
+    true = ets:insert(?RULE_TAB, {Id, Rule}),
     ok.
 
 do_delete_rule(#{id := Id} = Rule) ->
@@ -492,10 +496,10 @@ do_delete_rule(#{id := Id} = Rule) ->
     true = ets:delete(?RULE_TAB, Id),
     ok.
 
-do_update_rule_index(#{id := Id, from := From} = Rule) ->
+do_update_rule_index(#{id := Id, from := From}) ->
     ok = lists:foreach(
         fun(Topic) ->
-            true = emqx_topic_index:insert(Topic, Id, Rule, ?RULE_TOPIC_INDEX)
+            true = emqx_topic_index:insert(Topic, Id, [], ?RULE_TOPIC_INDEX)
         end,
         From
     ).
diff --git a/deploy/charts/emqx-enterprise/Chart.yaml b/deploy/charts/emqx-enterprise/Chart.yaml
index 575c6b354..b72b67e81 100644
--- a/deploy/charts/emqx-enterprise/Chart.yaml
+++ b/deploy/charts/emqx-enterprise/Chart.yaml
@@ -14,8 +14,8 @@ type: application
 
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
-version: 5.2.0-alpha.3
+version: 5.2.0-alpha.4
 
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application.
-appVersion: 5.2.0-alpha.3
+appVersion: 5.2.0-alpha.4
diff --git a/rel/i18n/emqx_schema.hocon b/rel/i18n/emqx_schema.hocon
index 6c21b31ea..9e33ffb57 100644
--- a/rel/i18n/emqx_schema.hocon
+++ b/rel/i18n/emqx_schema.hocon
@@ -1549,6 +1549,13 @@ fields_ws_opts_max_frame_size.label:
 sys_event_messages.desc:
 """Client events messages."""
 
+broker_routing_storage_schema.desc:
+"""Routing storage schema.
+Set <code>v1</code> to leave the default.
+<code>v2</code> is introduced in 5.2. It enables routing through 2 separate tables, one for topic filter and one for regular topic subscriptions. This schema should increase both subscription and routing performance at the cost of slight increase in memory consumption per subscription.
+NOTE: Schema <code>v2</code> is still experimental.
+NOTE: Full non-rolling cluster restart is needed after altering this option for it to take any effect."""
+
 broker_perf_trie_compaction.desc:
 """Enable trie path compaction.
 Enabling it significantly improves wildcard topic subscribe rate, if wildcard topics have unique prefixes like: 'sensor/{{id}}/+/', where ID is unique per subscriber.