feat: authz & authn mysql resource support prepare sql query

apps/emqx_authn/src/simple_authn/emqx_authn_mysql.erl

@@ -89,12 +89,12 @@ create(
     } = Config
 ) ->
     ok = emqx_authn_password_hashing:init(Algorithm),
-    {Query, PlaceHolders} = emqx_authn_utils:parse_sql(Query0, '?'),
+    {PrepareSqlKey, PrepareStatement} = emqx_authn_utils:parse_sql(Query0, '?'),
     ResourceId = emqx_authn_utils:make_resource_id(?MODULE),
     State = #{
         password_hash_algorithm => Algorithm,
-        query => Query,
-        placeholders => PlaceHolders,
+        prepare_sql_key => PrepareSqlKey,
+        prepare_sql_statement => PrepareStatement,
         query_timeout => QueryTimeout,
         resource_id => ResourceId
     },
@@ -107,10 +107,14 @@ create(
             #{}
         )
     of
-        {ok, already_created} ->
-            {ok, State};
         {ok, _} ->
-            {ok, State};
+            case emqx_resource:query(ResourceId,
+                     {prepare_sql, [{PrepareSqlKey, PrepareStatement}]}) of
+                ok ->
+                    {ok, State};
+                {error, Reason} ->
+                    {error, Reason}
+            end;
         {error, Reason} ->
             {error, Reason}
     end.

apps/emqx_authz/src/emqx_authz_mysql.erl

@@ -52,17 +52,13 @@ init(#{query := SQL} = Source) ->
         {error, Reason} ->
             error({load_config_error, Reason});
         {ok, Id} ->
-            Source#{
-                annotations =>
-                    #{
-                        id => Id,
-                        query => emqx_authz_utils:parse_sql(
-                            SQL,
-                            '?',
-                            ?PLACEHOLDERS
-                        )
-                    }
-            }
+            {PrepareKey, PrepareStatement} = emqx_authz_utils:parse_sql(SQL, '?', ?PLACEHOLDERS),
+            case emqx_resource:query(Id, {prepare_sql, [{PrepareKey, PrepareStatement}]}) of
+                ok ->
+                    Source#{annotations => #{id => Id, prepare => {PrepareKey, PrepareStatement}}};
+                {error, Reason} ->
+                    error({load_config_error, Reason})
+            end
     end.
 
 destroy(#{annotations := #{id := Id}}) ->
@@ -75,12 +71,12 @@ authorize(
     #{
         annotations := #{
             id := ResourceID,
-            query := {Query, Params}
+            prepare := {PrepareKey, PrepareStatement}
         }
     }
 ) ->
-    RenderParams = emqx_authz_utils:render_sql_params(Params, Client),
-    case emqx_resource:query(ResourceID, {sql, Query, RenderParams}) of
+    RenderParams = emqx_authz_utils:render_sql_params(PrepareStatement, Client),
+    case emqx_resource:query(ResourceID, {sql, PrepareKey, RenderParams}) of
         {ok, _Columns, []} ->
             nomatch;
         {ok, Columns, Rows} ->
@@ -89,7 +85,7 @@ authorize(
             ?SLOG(error, #{
                 msg => "query_mysql_error",
                 reason => Reason,
-                query => Query,
+                prepare => {PrepareKey, PrepareStatement},
                 params => RenderParams,
                 resource_id => ResourceID
             }),

apps/emqx_connector/src/emqx_connector_mysql.erl

@@ -94,6 +94,9 @@ on_stop(InstId, #{poolname := PoolName}) ->
                   connector => InstId}),
     emqx_plugin_libs_pool:stop_pool(PoolName).
 
+on_query(_InstId, {prepare_sql, Prepares}, _AfterQuery, #{poolname := PoolName}) ->
+    prepare_sql(Prepares, PoolName);
+
 on_query(InstId, {Type, SQLOrKey}, AfterQuery, #{poolname := _PoolName} = State) ->
     on_query(InstId, {Type, SQLOrKey, [], default_timeout}, AfterQuery, State);
 on_query(InstId, {Type, SQLOrKey, Params}, AfterQuery, #{poolname := _PoolName} = State) ->