Merge pull request #13861 from terry-xiaoyu/undefined-vars-as-null

Undefined vars as null

apps/emqx_bridge/src/schema/emqx_bridge_v2_schema.erl

@@ -59,7 +59,8 @@
 -export([source_resource_opts_fields/0, source_resource_opts_fields/1]).
 
 -export([
-    api_fields/3
+    api_fields/3,
+    undefined_as_null_field/0
 ]).
 
 -export([
@@ -330,6 +331,16 @@ api_fields("post_source", Type, Fields) ->
 api_fields("put_source", _Type, Fields) ->
     Fields.
 
+undefined_as_null_field() ->
+    {undefined_vars_as_null,
+        ?HOCON(
+            boolean(),
+            #{
+                default => false,
+                desc => ?DESC("undefined_vars_as_null")
+            }
+        )}.
+
 %%======================================================================================
 %% HOCON Schema Callbacks
 %%======================================================================================

apps/emqx_bridge_clickhouse/src/emqx_bridge_clickhouse.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_clickhouse, [
     {description, "EMQX Enterprise ClickHouse Bridge"},
-    {vsn, "0.4.2"},
+    {vsn, "0.4.3"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_bridge_clickhouse/src/emqx_bridge_clickhouse.erl

@@ -129,6 +129,7 @@ fields(clickhouse_action) ->
 fields(action_parameters) ->
     [
         sql_field(),
+        emqx_bridge_v2_schema:undefined_as_null_field(),
         batch_value_separator_field()
     ];
 fields(connector_resource_opts) ->

apps/emqx_bridge_clickhouse/src/emqx_bridge_clickhouse_connector.erl

@@ -294,8 +294,10 @@ on_stop(InstanceID, _State) ->
 %% channel related emqx_resouce callbacks
 %% -------------------------------------------------------------------
 on_add_channel(_InstId, #{channels := Channs} = OldState, ChannId, ChannConf0) ->
-    #{parameters := ParamConf} = ChannConf0,
-    NewChanns = Channs#{ChannId => #{templates => prepare_sql_templates(ParamConf)}},
+    #{parameters := ChannelConf} = ChannConf0,
+    NewChanns = Channs#{
+        ChannId => #{templates => prepare_sql_templates(ChannelConf), channel_conf => ChannelConf}
+    },
     {ok, OldState#{channels => NewChanns}}.
 
 on_remove_channel(_InstanceId, #{channels := Channels} = State, ChannId) ->
@@ -387,22 +389,28 @@ on_query(
     }),
     %% Have we got a query or data to fit into an SQL template?
     SimplifiedRequestType = query_type(RequestType),
+    ChannelState = get_channel_state(RequestType, State),
     Templates = get_templates(RequestType, State),
-    SQL = get_sql(SimplifiedRequestType, Templates, DataOrSQL),
+    SQL = get_sql(
+        SimplifiedRequestType, Templates, DataOrSQL, maps:get(channel_conf, ChannelState, #{})
+    ),
     ClickhouseResult = execute_sql_in_clickhouse_server(RequestType, PoolName, SQL),
     transform_and_log_clickhouse_result(ClickhouseResult, ResourceID, SQL).
 
 get_templates(ChannId, State) ->
+    maps:get(templates, get_channel_state(ChannId, State), #{}).
+
+get_channel_state(ChannId, State) ->
     case maps:find(channels, State) of
         {ok, Channels} ->
-            maps:get(templates, maps:get(ChannId, Channels, #{}), #{});
+            maps:get(ChannId, Channels, #{});
         error ->
             #{}
     end.
 
-get_sql(channel_message, #{send_message_template := PreparedSQL}, Data) ->
-    emqx_placeholder:proc_tmpl(PreparedSQL, Data, #{return => full_binary});
-get_sql(_, _, SQL) ->
+get_sql(channel_message, #{send_message_template := PreparedSQL}, Data, ChannelConf) ->
+    proc_nullable_tmpl(PreparedSQL, Data, ChannelConf);
+get_sql(_, _, SQL, _) ->
     SQL.
 
 query_type(sql) ->
@@ -425,8 +433,9 @@ on_batch_query(ResourceID, BatchReq, #{pool_name := PoolName} = State) ->
     {[ChannId | _] = Keys, ObjectsToInsert} = lists:unzip(BatchReq),
     ensure_channel_messages(Keys),
     Templates = get_templates(ChannId, State),
+    ChannelState = get_channel_state(ChannId, State),
     %% Create batch insert SQL statement
-    SQL = objects_to_sql(ObjectsToInsert, Templates),
+    SQL = objects_to_sql(ObjectsToInsert, Templates, maps:get(channel_conf, ChannelState, #{})),
     %% Do the actual query in the database
     ResultFromClickhouse = execute_sql_in_clickhouse_server(ChannId, PoolName, SQL),
     %% Transform the result to a better format
@@ -447,20 +456,26 @@ objects_to_sql(
     #{
         send_message_template := InsertTemplate,
         extend_send_message_template := BulkExtendInsertTemplate
-    }
+    },
+    ChannelConf
 ) ->
     %% Prepare INSERT-statement and the first row after VALUES
-    InsertStatementHead = emqx_placeholder:proc_tmpl(InsertTemplate, FirstObject),
+    InsertStatementHead = proc_nullable_tmpl(InsertTemplate, FirstObject, ChannelConf),
     FormatObjectDataFunction =
         fun(Object) ->
-            emqx_placeholder:proc_tmpl(BulkExtendInsertTemplate, Object)
+            proc_nullable_tmpl(BulkExtendInsertTemplate, Object, ChannelConf)
         end,
     InsertStatementTail = lists:map(FormatObjectDataFunction, RemainingObjects),
     CompleteStatement = erlang:iolist_to_binary([InsertStatementHead, InsertStatementTail]),
     CompleteStatement;
-objects_to_sql(_, _) ->
+objects_to_sql(_, _, _) ->
     erlang:error(<<"Templates for bulk insert missing.">>).
 
+proc_nullable_tmpl(Template, Data, #{undefined_vars_as_null := true}) ->
+    emqx_placeholder:proc_nullable_tmpl(Template, Data);
+proc_nullable_tmpl(Template, Data, _) ->
+    emqx_placeholder:proc_tmpl(Template, Data).
+
 %% -------------------------------------------------------------------
 %% Helper functions that are used by both on_query/3 and on_batch_query/3
 %% -------------------------------------------------------------------

apps/emqx_bridge_clickhouse/test/emqx_bridge_clickhouse_SUITE.erl

@@ -9,6 +9,7 @@
 
 -define(CLICKHOUSE_HOST, "clickhouse").
 -define(CLICKHOUSE_PORT, "8123").
+-include_lib("eunit/include/eunit.hrl").
 -include_lib("common_test/include/ct.hrl").
 -include_lib("emqx_connector/include/emqx_connector.hrl").
 
@@ -177,9 +178,12 @@ parse_and_check(ConfigString, BridgeType, Name) ->
     RetConfig.
 
 make_bridge(Config) ->
+    make_bridge(Config, #{}).
+
+make_bridge(Config, Overrides) ->
     Type = <<"clickhouse">>,
     Name = atom_to_binary(?MODULE),
-    BridgeConfig = clickhouse_config(Config),
+    BridgeConfig = maps:merge(clickhouse_config(Config), Overrides),
     {ok, _} = emqx_bridge:create(
         Type,
         Name,
@@ -252,6 +256,21 @@ t_send_message_query(Config) ->
     delete_bridge(),
     ok.
 
+t_undefined_vars_as_null(Config) ->
+    BridgeID = make_bridge(#{enable_batch => false}, #{<<"undefined_vars_as_null">> => true}),
+    Key = 42,
+    Payload = #{key => Key, data => undefined, timestamp => 10000},
+    %% This will use the SQL template included in the bridge
+    emqx_bridge:send_message(BridgeID, Payload),
+    %% Check that the data got to the database
+    check_key_in_clickhouse(Key, Config),
+    ClickhouseConnection = proplists:get_value(clickhouse_connection, Config),
+    SQL = io_lib:format("SELECT data FROM mqtt.mqtt_test WHERE key = ~p", [Key]),
+    {ok, 200, ResultString} = clickhouse:query(ClickhouseConnection, SQL, []),
+    ?assertMatch(<<"null">>, iolist_to_binary(string:trim(ResultString))),
+    delete_bridge(),
+    ok.
+
 t_send_simple_batch(Config) ->
     send_simple_batch_helper(Config, #{}).
 

apps/emqx_bridge_dynamo/src/emqx_bridge_dynamo.erl

@@ -200,7 +200,8 @@ fields("config_connector") ->
         end,
         Config,
         [
-            table
+            table,
+            undefined_vars_as_null
         ]
     );
 fields(connector_resource_opts) ->

apps/emqx_bridge_dynamo/src/emqx_bridge_dynamo_connector.erl

@@ -63,7 +63,8 @@ fields(config) ->
                 }
             )},
         {pool_size, fun emqx_connector_schema_lib:pool_size/1},
-        {auto_reconnect, fun emqx_connector_schema_lib:auto_reconnect/1}
+        {auto_reconnect, fun emqx_connector_schema_lib:auto_reconnect/1},
+        emqx_bridge_v2_schema:undefined_as_null_field()
     ].
 
 %%========================================================================================
@@ -253,7 +254,7 @@ do_query(
                 ecpool:pick_and_do(
                     PoolName,
                     {emqx_bridge_dynamo_connector_client, query, [
-                        Table, QueryTuple, Templates, TraceRenderedCTX
+                        Table, QueryTuple, Templates, TraceRenderedCTX, ChannelState
                     ]},
                     no_handover
                 );

apps/emqx_bridge_dynamo/src/emqx_bridge_dynamo_connector_client.erl

@@ -10,7 +10,7 @@
 -export([
     start_link/1,
     is_connected/2,
-    query/5
+    query/6
 ]).
 
 %% gen_server callbacks
@@ -24,7 +24,7 @@
 ]).
 
 -ifdef(TEST).
--export([execute/2]).
+-export([execute/3]).
 -endif.
 
 -include_lib("emqx/include/emqx_trace.hrl").
@@ -42,8 +42,10 @@ is_connected(Pid, Timeout) ->
             {false, Error}
     end.
 
-query(Pid, Table, Query, Templates, TraceRenderedCTX) ->
-    gen_server:call(Pid, {query, Table, Query, Templates, TraceRenderedCTX}, infinity).
+query(Pid, Table, Query, Templates, TraceRenderedCTX, ChannelState) ->
+    gen_server:call(
+        Pid, {query, Table, Query, Templates, TraceRenderedCTX, ChannelState}, infinity
+    ).
 
 %%--------------------------------------------------------------------
 %% @doc
@@ -79,14 +81,14 @@ handle_call(is_connected, _From, State) ->
                 {false, Error}
         end,
     {reply, IsConnected, State};
-handle_call({query, Table, Query, Templates, TraceRenderedCTX}, _From, State) ->
-    Result = do_query(Table, Query, Templates, TraceRenderedCTX),
+handle_call({query, Table, Query, Templates, TraceRenderedCTX, ChannelState}, _From, State) ->
+    Result = do_query(Table, Query, Templates, TraceRenderedCTX, ChannelState),
     {reply, Result, State};
 handle_call(_Request, _From, State) ->
     {reply, ok, State}.
 
-handle_cast({query, Table, Query, Templates, {ReplyFun, [Context]}}, State) ->
-    Result = do_query(Table, Query, Templates, {fun(_, _) -> ok end, none}),
+handle_cast({query, Table, Query, Templates, {ReplyFun, [Context]}, ChannelState}, State) ->
+    Result = do_query(Table, Query, Templates, {fun(_, _) -> ok end, none}, ChannelState),
     ReplyFun(Context, Result),
     {noreply, State};
 handle_cast(_Request, State) ->
@@ -104,9 +106,9 @@ code_change(_OldVsn, State, _Extra) ->
 %%%===================================================================
 %%% Internal functions
 %%%===================================================================
-do_query(Table, Query0, Templates, TraceRenderedCTX) ->
+do_query(Table, Query0, Templates, TraceRenderedCTX, ChannelState) ->
     try
-        Query = apply_template(Query0, Templates),
+        Query = apply_template(Query0, Templates, ChannelState),
         emqx_trace:rendered_action_template_with_ctx(TraceRenderedCTX, #{
             table => Table,
             query => #emqx_trace_format_func_data{
@@ -114,12 +116,12 @@ do_query(Table, Query0, Templates, TraceRenderedCTX) ->
                 data = Query
             }
         }),
-        execute(Query, Table)
+        execute(Query, Table, ChannelState)
     catch
         error:{unrecoverable_error, Reason} ->
             {error, {unrecoverable_error, Reason}};
-        _Type:Reason ->
-            {error, {unrecoverable_error, {invalid_request, Reason}}}
+        Err:Reason:ST ->
+            {error, {unrecoverable_error, {invalid_request, {Err, Reason, ST}}}}
     end.
 
 trace_format_query({Type, Data}) ->
@@ -131,68 +133,76 @@ trace_format_query(Query) ->
     Query.
 
 %% some simple query commands for authn/authz or test
-execute({insert_item, Msg}, Table) ->
-    Item = convert_to_item(Msg),
+execute({insert_item, Msg}, Table, ChannelState) ->
+    Item = convert_to_item(Msg, ChannelState),
     erlcloud_ddb2:put_item(Table, Item);
-execute({delete_item, Key}, Table) ->
+execute({delete_item, Key}, Table, _) ->
     erlcloud_ddb2:delete_item(Table, Key);
-execute({get_item, Key}, Table) ->
+execute({get_item, Key}, Table, _) ->
     erlcloud_ddb2:get_item(Table, Key);
 %% commands for data bridge query or batch query
-execute({send_message, Msg}, Table) ->
-    Item = convert_to_item(Msg),
+execute({send_message, Msg}, Table, ChannelState) ->
+    Item = convert_to_item(Msg, ChannelState),
     erlcloud_ddb2:put_item(Table, Item);
-execute([{put, _} | _] = Msgs, Table) ->
+execute([{put, _} | _] = Msgs, Table, _) ->
     %% type of batch_write_item argument :: batch_write_item_request_items()
     %% batch_write_item_request_items() :: maybe_list(batch_write_item_request_item())
     %% batch_write_item_request_item() :: {table_name(), list(batch_write_item_request())}
     %% batch_write_item_request() :: {put, item()} | {delete, key()}
     erlcloud_ddb2:batch_write_item({Table, Msgs}).
 
-apply_template({Key, Msg} = Req, Templates) ->
-    case maps:get(Key, Templates, undefined) of
-        undefined ->
-            Req;
-        Template ->
-            {Key, emqx_placeholder:proc_tmpl(Template, Msg)}
+apply_template({Key, Msg} = Req, Templates, _) ->
+    case maps:find(Key, Templates) of
+        error -> Req;
+        {ok, Template} -> {Key, emqx_placeholder:proc_tmpl(Template, Msg)}
     end;
 %% now there is no batch delete, so
 %% 1. we can simply replace the `send_message` to `put`
 %% 2. convert the message to in_item() here, not at the time when calling `batch_write_items`,
 %%    so we can reduce some list map cost
-apply_template([{_, _Msg} | _] = Msgs, Templates) ->
+apply_template([{_, _Msg} | _] = Msgs, Templates, ChannelState) ->
     lists:map(
         fun(Req) ->
-            {_, Msg} = apply_template(Req, Templates),
-            {put, convert_to_item(Msg)}
+            {_, Msg} = apply_template(Req, Templates, ChannelState),
+            {put, convert_to_item(Msg, ChannelState)}
         end,
         Msgs
     ).
 
-convert_to_item(Msg) when is_map(Msg), map_size(Msg) > 0 ->
+convert_to_item(Msg, ChannelState) when is_map(Msg), map_size(Msg) > 0 ->
     maps:fold(
         fun
             (_K, <<>>, AccIn) ->
                 AccIn;
             (K, V, AccIn) ->
-                [{convert2binary(K), convert2binary(V)} | AccIn]
+                [{to_bin(K), val_to_bin(V, ChannelState)} | AccIn]
         end,
         [],
         Msg
     );
-convert_to_item(MsgBin) when is_binary(MsgBin) ->
+convert_to_item(MsgBin, ChannelState) when is_binary(MsgBin) ->
     Msg = emqx_utils_json:decode(MsgBin),
-    convert_to_item(Msg);
-convert_to_item(Item) ->
+    convert_to_item(Msg, ChannelState);
+convert_to_item(Item, _) ->
     erlang:throw({invalid_item, Item}).
 
-convert2binary(Value) when is_atom(Value) ->
+val_to_bin(Null, #{undefined_vars_as_null := true}) when
+    Null =:= <<"undefined">>;
+    Null =:= <<"null">>;
+    Null =:= undefined;
+    Null =:= null
+->
+    {null, true};
+val_to_bin(Val, _) ->
+    to_bin(Val).
+
+to_bin(Value) when is_atom(Value) ->
     erlang:atom_to_binary(Value, utf8);
-convert2binary(Value) when is_binary(Value); is_number(Value) ->
+to_bin(Value) when is_binary(Value); is_number(Value) ->
     Value;
-convert2binary(Value) when is_list(Value) ->
+to_bin(Value) when is_list(Value) ->
     unicode:characters_to_binary(Value);
-convert2binary(Value) when is_map(Value) ->
+to_bin(Value) when is_map(Value) ->
     emqx_utils_json:encode(Value).
 
 to_str(List) when is_list(List) ->

apps/emqx_bridge_dynamo/test/emqx_bridge_dynamo_SUITE.erl

@@ -348,12 +348,15 @@ directly_setup_dynamo() ->
 
 directly_query(Query) ->
     directly_setup_dynamo(),
-    emqx_bridge_dynamo_connector_client:execute(Query, ?TABLE_BIN).
+    emqx_bridge_dynamo_connector_client:execute(Query, ?TABLE_BIN, #{}).
 
 directly_get_payload(Key) ->
+    directly_get_field(Key, <<"payload">>).
+
+directly_get_field(Key, Field) ->
     case directly_query({get_item, {<<"id">>, Key}}) of
         {ok, Values} ->
-            proplists:get_value(<<"payload">>, Values, {error, {invalid_item, Values}});
+            proplists:get_value(Field, Values, {error, {invalid_item, Values}});
         Error ->
             Error
     end.
@@ -370,7 +373,7 @@ t_setup_via_config_and_publish(Config) ->
         create_bridge(Config)
     ),
     MsgId = emqx_utils:gen_id(),
-    SentData = #{clientid => <<"clientid">>, id => MsgId, payload => ?PAYLOAD},
+    SentData = #{clientid => <<"clientid">>, id => MsgId, payload => ?PAYLOAD, foo => undefined},
     ?check_trace(
         begin
             ?wait_async_action(
@@ -384,6 +387,43 @@ t_setup_via_config_and_publish(Config) ->
                 ?PAYLOAD,
                 directly_get_payload(MsgId)
             ),
+            ?assertMatch(
+                %% the old behavior without undefined_vars_as_null
+                <<"undefined">>,
+                directly_get_field(MsgId, <<"foo">>)
+            ),
+            ok
+        end,
+        fun(Trace0) ->
+            Trace = ?of_kind(dynamo_connector_query_return, Trace0),
+            ?assertMatch([#{result := {ok, _}}], Trace),
+            ok
+        end
+    ),
+    ok.
+
+t_undefined_vars_as_null(Config) ->
+    ?assertNotEqual(undefined, get(aws_config)),
+    create_table(Config),
+    ?assertMatch(
+        {ok, _},
+        create_bridge(Config, #{<<"undefined_vars_as_null">> => true})
+    ),
+    MsgId = emqx_utils:gen_id(),
+    SentData = #{clientid => <<"clientid">>, id => MsgId, payload => undefined},
+    ?check_trace(
+        begin
+            ?wait_async_action(
+                ?assertMatch(
+                    {ok, _}, send_message(Config, SentData)
+                ),
+                #{?snk_kind := dynamo_connector_query_return},
+                10_000
+            ),
+            ?assertMatch(
+                undefined,
+                directly_get_payload(MsgId)
+            ),
             ok
         end,
         fun(Trace0) ->

apps/emqx_bridge_mysql/src/emqx_bridge_mysql.erl

@@ -148,7 +148,8 @@ fields(action_parameters) ->
             mk(
                 emqx_schema:template(),
                 #{desc => ?DESC("sql_template"), default => ?DEFAULT_SQL, format => <<"sql">>}
-            )}
+            )},
+        emqx_bridge_v2_schema:undefined_as_null_field()
     ];
 fields("config_connector") ->
     emqx_connector_schema:common_fields() ++

apps/emqx_bridge_mysql/src/emqx_bridge_mysql_connector.erl

@@ -130,12 +130,13 @@ on_batch_query(
     Result = emqx_mysql:on_batch_query(
         InstanceId,
         BatchRequest,
-        MergedState1
+        MergedState1,
+        ChannelConfig
     ),
     ?tp(mysql_connector_on_batch_query_return, #{instance_id => InstanceId, result => Result}),
     Result;
 on_batch_query(InstanceId, BatchRequest, _State = #{connector_state := ConnectorState}) ->
-    emqx_mysql:on_batch_query(InstanceId, BatchRequest, ConnectorState).
+    emqx_mysql:on_batch_query(InstanceId, BatchRequest, ConnectorState, #{}).
 
 on_remove_channel(
     _InstanceId, #{channels := Channels, connector_state := ConnectorState} = State, ChannelId

apps/emqx_bridge_mysql/test/emqx_bridge_mysql_SUITE.erl

@@ -426,6 +426,33 @@ t_setup_via_config_and_publish(Config) ->
     ),
     ok.
 
+t_undefined_vars_as_null(Config) ->
+    ?assertMatch(
+        {ok, _},
+        create_bridge(Config, #{<<"undefined_vars_as_null">> => true})
+    ),
+    SentData = #{payload => undefined, timestamp => 1668602148000},
+    ?check_trace(
+        begin
+            ?wait_async_action(
+                ?assertEqual(ok, send_message(Config, SentData)),
+                #{?snk_kind := mysql_connector_query_return},
+                10_000
+            ),
+            ?assertMatch(
+                {ok, [<<"payload">>], [[null]]},
+                connect_and_get_payload(Config)
+            ),
+            ok
+        end,
+        fun(Trace0) ->
+            Trace = ?of_kind(mysql_connector_query_return, Trace0),
+            ?assertMatch([#{result := ok}], Trace),
+            ok
+        end
+    ),
+    ok.
+
 t_setup_via_http_api_and_publish(Config) ->
     BridgeType = ?config(mysql_bridge_type, Config),
     Name = ?config(mysql_name, Config),

apps/emqx_bridge_sqlserver/src/emqx_bridge_sqlserver.erl

@@ -171,7 +171,8 @@ fields("config") ->
                     default => #{},
                     desc => ?DESC(emqx_resource_schema, <<"resource_opts">>)
                 }
-            )}
+            )},
+        emqx_bridge_v2_schema:undefined_as_null_field()
     ] ++ driver_fields() ++
         (emqx_bridge_sqlserver_connector:fields(config) --
             emqx_connector_schema_lib:prepare_statement_fields());
@@ -194,7 +195,8 @@ fields(action_parameters) ->
             mk(
                 emqx_schema:template(),
                 #{desc => ?DESC("sql_template"), default => ?DEFAULT_SQL, format => <<"sql">>}
-            )}
+            )},
+        emqx_bridge_v2_schema:undefined_as_null_field()
     ];
 fields("creation_opts") ->
     emqx_resource_schema:fields("creation_opts");

apps/emqx_bridge_sqlserver/src/emqx_bridge_sqlserver_connector.erl

@@ -248,9 +248,9 @@ on_add_channel(
     {ok, NewState}.
 
 create_channel_state(
-    #{parameters := Conf} = _ChannelConfig
+    #{parameters := ChannelConf}
 ) ->
-    State = #{sql_templates => parse_sql_template(Conf)},
+    State = #{sql_templates => parse_sql_template(ChannelConf), channel_conf => ChannelConf},
     {ok, State}.
 
 on_remove_channel(
@@ -414,10 +414,10 @@ do_query(
 
     ChannelId = get_channel_id(Query),
     QueryTuple = get_query_tuple(Query),
-    #{sql_templates := Templates} = _ChannelState = maps:get(ChannelId, Channels),
-
+    #{sql_templates := Templates} = ChannelState = maps:get(ChannelId, Channels),
+    ChannelConf = maps:get(channel_conf, ChannelState, #{}),
     %% only insert sql statement for single query and batch query
-    case apply_template(QueryTuple, Templates) of
+    case apply_template(QueryTuple, Templates, ChannelConf) of
         {?ACTION_SEND_MESSAGE, SQL} ->
             emqx_trace:rendered_action_template(ChannelId, #{
                 sql => SQL
@@ -560,36 +560,42 @@ parse_sql_template([], BatchInsertTks) ->
 
 %% single insert
 apply_template(
-    {?ACTION_SEND_MESSAGE = _Key, _Msg} = Query, Templates
+    {?ACTION_SEND_MESSAGE = _Key, _Msg} = Query, Templates, ChannelConf
 ) ->
     %% TODO: fix emqx_placeholder:proc_tmpl/2
     %% it won't add single quotes for string
-    apply_template([Query], Templates);
+    apply_template([Query], Templates, ChannelConf);
 %% batch inserts
 apply_template(
     [{?ACTION_SEND_MESSAGE = Key, _Msg} | _T] = BatchReqs,
-    #{?BATCH_INSERT_TEMP := BatchInsertsTks} = _Templates
+    #{?BATCH_INSERT_TEMP := BatchInsertsTks} = _Templates,
+    ChannelConf
 ) ->
     case maps:get(Key, BatchInsertsTks, undefined) of
         undefined ->
             BatchReqs;
         #{?BATCH_INSERT_PART := BatchInserts, ?BATCH_PARAMS_TOKENS := BatchParamsTks} ->
-            SQL = proc_batch_sql(BatchReqs, BatchInserts, BatchParamsTks),
+            SQL = proc_batch_sql(BatchReqs, BatchInserts, BatchParamsTks, ChannelConf),
             {Key, SQL}
     end;
-apply_template(Query, Templates) ->
+apply_template(Query, Templates, _) ->
     %% TODO: more detail information
     ?SLOG(error, #{msg => "apply_sql_template_failed", query => Query, templates => Templates}),
     {error, failed_to_apply_sql_template}.
 
-proc_batch_sql(BatchReqs, BatchInserts, Tokens) ->
+proc_batch_sql(BatchReqs, BatchInserts, Tokens, ChannelConf) ->
     Values = erlang:iolist_to_binary(
         lists:join($,, [
-            emqx_placeholder:proc_sql_param_str(Tokens, Msg)
+            proc_msg(Tokens, Msg, ChannelConf)
          || {_, Msg} <- BatchReqs
         ])
     ),
     <<BatchInserts/binary, " values ", Values/binary>>.
 
+proc_msg(Tokens, Msg, #{undefined_vars_as_null := true}) ->
+    emqx_placeholder:proc_sql_param_str2(Tokens, Msg);
+proc_msg(Tokens, Msg, _) ->
+    emqx_placeholder:proc_sql_param_str(Tokens, Msg).
+
 to_bin(List) when is_list(List) ->
     unicode:characters_to_binary(List, utf8).

apps/emqx_bridge_sqlserver/test/emqx_bridge_sqlserver_SUITE.erl

@@ -189,6 +189,33 @@ t_setup_via_config_and_publish(Config) ->
     ),
     ok.
 
+t_undefined_vars_as_null(Config) ->
+    ?assertMatch(
+        {ok, _},
+        create_bridge(Config, #{<<"undefined_vars_as_null">> => true})
+    ),
+    SentData = maps:put(payload, undefined, sent_data("tmp")),
+    ?check_trace(
+        begin
+            ?wait_async_action(
+                ?assertEqual(ok, send_message(Config, SentData)),
+                #{?snk_kind := sqlserver_connector_query_return},
+                10_000
+            ),
+            ?assertMatch(
+                [{null}],
+                connect_and_get_payload(Config)
+            ),
+            ok
+        end,
+        fun(Trace0) ->
+            Trace = ?of_kind(sqlserver_connector_query_return, Trace0),
+            ?assertMatch([#{result := ok}], Trace),
+            ok
+        end
+    ),
+    ok.
+
 t_setup_via_http_api_and_publish(Config) ->
     BridgeType = ?config(sqlserver_bridge_type, Config),
     Name = ?config(sqlserver_name, Config),

apps/emqx_bridge_tdengine/src/emqx_bridge_tdengine.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_tdengine, [
     {description, "EMQX Enterprise TDEngine Bridge"},
-    {vsn, "0.2.2"},
+    {vsn, "0.2.3"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_bridge_tdengine/src/emqx_bridge_tdengine.erl

@@ -90,6 +90,7 @@ fields("config") ->
                     format => <<"sql">>
                 }
             )},
+        emqx_bridge_v2_schema:undefined_as_null_field(),
         {local_topic, mk(binary(), #{desc => ?DESC("local_topic"), default => undefined})}
     ] ++
         emqx_resource_schema:fields("resource_opts") ++
@@ -131,7 +132,8 @@ fields(action_parameters) ->
                     default => ?DEFAULT_SQL,
                     format => <<"sql">>
                 }
-            )}
+            )},
+        emqx_bridge_v2_schema:undefined_as_null_field()
     ];
 fields("post_bridge_v2") ->
     emqx_bridge_schema:type_and_name_fields(enum([tdengine])) ++ fields(action_config);

apps/emqx_bridge_tdengine/src/emqx_bridge_tdengine_connector.erl

@@ -35,7 +35,7 @@
 
 -export([connector_examples/1]).
 
--export([connect/1, do_get_status/1, execute/3, do_batch_insert/5]).
+-export([connect/1, do_get_status/1, execute/3, do_batch_insert/6]).
 
 -import(hoconsc, [mk/2, enum/1, ref/2]).
 
@@ -188,8 +188,8 @@ on_stop(InstanceId, _State) ->
 
 on_query(InstanceId, {ChannelId, Data}, #{channels := Channels} = State) ->
     case maps:find(ChannelId, Channels) of
-        {ok, #{insert := Tokens, opts := Opts}} ->
-            Query = emqx_placeholder:proc_tmpl(Tokens, Data),
+        {ok, #{insert := Tokens, opts := Opts} = ChannelState} ->
+            Query = proc_nullable_tmpl(Tokens, Data, maps:get(channel_conf, ChannelState, #{})),
             emqx_trace:rendered_action_template(ChannelId, #{query => Query}),
             do_query_job(InstanceId, {?MODULE, execute, [Query, Opts]}, State);
         _ ->
@@ -203,11 +203,12 @@ on_batch_query(
     #{channels := Channels} = State
 ) ->
     case maps:find(ChannelId, Channels) of
-        {ok, #{batch := Tokens, opts := Opts}} ->
+        {ok, #{batch := Tokens, opts := Opts} = ChannelState} ->
             TraceRenderedCTX = emqx_trace:make_rendered_action_template_trace_context(ChannelId),
+            ChannelConf = maps:get(channel_conf, ChannelState, #{}),
             do_query_job(
                 InstanceId,
-                {?MODULE, do_batch_insert, [Tokens, BatchReq, Opts, TraceRenderedCTX]},
+                {?MODULE, do_batch_insert, [Tokens, BatchReq, Opts, TraceRenderedCTX, ChannelConf]},
                 State
             );
         _ ->
@@ -273,7 +274,7 @@ on_add_channel(
     #{channels := Channels} = OldState,
     ChannelId,
     #{
-        parameters := #{database := Database, sql := SQL}
+        parameters := #{database := Database, sql := SQL} = ChannelConf
     }
 ) ->
     case maps:is_key(ChannelId, Channels) of
@@ -283,7 +284,8 @@ on_add_channel(
             case parse_prepare_sql(SQL) of
                 {ok, Result} ->
                     Opts = [{db_name, Database}],
-                    Channels2 = Channels#{ChannelId => Result#{opts => Opts}},
+                    Channel = Result#{opts => Opts, channel_conf => ChannelConf},
+                    Channels2 = Channels#{ChannelId => Channel},
                     {ok, OldState#{channels := Channels2}};
                 Error ->
                     Error
@@ -349,8 +351,8 @@ do_query_job(InstanceId, Job, #{pool_name := PoolName} = State) ->
 execute(Conn, Query, Opts) ->
     tdengine:insert(Conn, Query, Opts).
 
-do_batch_insert(Conn, Tokens, BatchReqs, Opts, TraceRenderedCTX) ->
-    SQL = aggregate_query(Tokens, BatchReqs, <<"INSERT INTO">>),
+do_batch_insert(Conn, Tokens, BatchReqs, Opts, TraceRenderedCTX, ChannelConf) ->
+    SQL = aggregate_query(Tokens, BatchReqs, <<"INSERT INTO">>, ChannelConf),
     try
         emqx_trace:rendered_action_template_with_ctx(
             TraceRenderedCTX,
@@ -362,16 +364,21 @@ do_batch_insert(Conn, Tokens, BatchReqs, Opts, TraceRenderedCTX) ->
             {error, Reason}
     end.
 
-aggregate_query(BatchTks, BatchReqs, Acc) ->
+aggregate_query(BatchTks, BatchReqs, Acc, ChannelConf) ->
     lists:foldl(
         fun({_, Data}, InAcc) ->
-            InsertPart = emqx_placeholder:proc_tmpl(BatchTks, Data),
+            InsertPart = proc_nullable_tmpl(BatchTks, Data, ChannelConf),
             <<InAcc/binary, " ", InsertPart/binary>>
         end,
         Acc,
         BatchReqs
     ).
 
+proc_nullable_tmpl(Template, Data, #{undefined_vars_as_null := true}) ->
+    emqx_placeholder:proc_nullable_tmpl(Template, Data);
+proc_nullable_tmpl(Template, Data, _) ->
+    emqx_placeholder:proc_tmpl(Template, Data).
+
 connect(Opts) ->
     %% TODO: teach `tdengine` to accept 0-arity closures as passwords.
     {value, {password, Secret}, OptsRest} = lists:keytake(password, 1, Opts),

apps/emqx_bridge_tdengine/test/emqx_bridge_tdengine_SUITE.erl

@@ -27,7 +27,6 @@
 -define(SQL_DROP_TABLE, "DROP TABLE t_mqtt_msg").
 -define(SQL_DROP_STABLE, "DROP STABLE s_tab").
 -define(SQL_DELETE, "DELETE FROM t_mqtt_msg").
--define(SQL_SELECT, "SELECT payload FROM t_mqtt_msg").
 
 -define(AUTO_CREATE_BRIDGE,
     "insert into ${clientid} USING s_tab TAGS ('${clientid}') values (${timestamp}, '${payload}')"
@@ -77,23 +76,23 @@ groups() ->
         {without_batch, TCs -- MustBatchCases}
     ].
 
+-define(APPS, [
+    emqx,
+    emqx_conf,
+    emqx_bridge_tdengine,
+    emqx_connector,
+    emqx_bridge,
+    emqx_rule_engine,
+    emqx_management
+]).
+
 init_per_suite(Config) ->
     emqx_bridge_v2_testlib:init_per_suite(
-        Config,
-        [
-            emqx,
-            emqx_conf,
-            emqx_bridge_tdengine,
-            emqx_connector,
-            emqx_bridge,
-            emqx_rule_engine,
-            emqx_management,
-            emqx_mgmt_api_test_util:emqx_dashboard()
-        ]
+        Config, ?APPS ++ [emqx_mgmt_api_test_util:emqx_dashboard()]
     ).
 
 end_per_suite(Config) ->
-    emqx_bridge_v2_testlib:end_per_suite(Config).
+    emqx_bridge_v2_testlib:end_per_suite([{apps, ?APPS} | Config]).
 
 init_per_group(async, Config) ->
     [{query_mode, async} | Config];
@@ -304,8 +303,11 @@ connect_and_clear_table(Config) ->
     ?WITH_CON({ok, _} = directly_query(Con, ?SQL_DELETE)).
 
 connect_and_get_payload(Config) ->
+    connect_and_get_column(Config, "SELECT payload FROM t_mqtt_msg").
+
+connect_and_get_column(Config, Select) ->
     ?WITH_CON(
-        {ok, #{<<"code">> := 0, <<"data">> := Result}} = directly_query(Con, ?SQL_SELECT)
+        {ok, #{<<"code">> := 0, <<"data">> := Result}} = directly_query(Con, Select)
     ),
     Result.
 
@@ -376,6 +378,41 @@ t_simple_insert(Config) ->
         connect_and_get_payload(Config)
     ).
 
+t_simple_insert_undefined(Config) ->
+    connect_and_clear_table(Config),
+
+    MakeMessageFun = fun() ->
+        #{payload => undefined, timestamp => 1668602148000, second_ts => 1668602148010}
+    end,
+
+    ok = emqx_bridge_v2_testlib:t_sync_query(
+        Config, MakeMessageFun, fun is_success_check/1, tdengine_connector_query_return
+    ),
+
+    ?assertMatch(
+        %% the old behavior without undefined_vars_as_null
+        [[<<"undefined">>], [<<"undefined">>]],
+        connect_and_get_payload(Config)
+    ).
+
+t_undefined_vars_as_null(Config0) ->
+    Config = patch_bridge_config(Config0, #{
+        <<"parameters">> => #{<<"undefined_vars_as_null">> => true}
+    }),
+    connect_and_clear_table(Config),
+
+    MakeMessageFun = fun() ->
+        #{payload => undefined, timestamp => 1668602148000, second_ts => 1668602148010}
+    end,
+    ok = emqx_bridge_v2_testlib:t_sync_query(
+        Config, MakeMessageFun, fun is_success_check/1, tdengine_connector_query_return
+    ),
+
+    ?assertMatch(
+        [[<<"null">>], [<<"null">>]],
+        connect_and_get_payload(Config)
+    ).
+
 t_batch_insert(Config) ->
     connect_and_clear_table(Config),
     ?assertMatch({ok, _}, emqx_bridge_v2_testlib:create_bridge(Config)),
@@ -503,3 +540,8 @@ t_auto_create_batch_insert(Config) ->
         end,
         [ClientId1, ClientId2, "test_" ++ ClientId1, "test_" ++ ClientId2]
     ).
+
+patch_bridge_config(Config, Overrides) ->
+    BridgeConfig0 = ?config(bridge_config, Config),
+    BridgeConfig1 = emqx_utils_maps:deep_merge(BridgeConfig0, Overrides),
+    [{bridge_config, BridgeConfig1} | proplists:delete(bridge_config, Config)].

apps/emqx_mysql/src/emqx_mysql.erl

@@ -30,7 +30,7 @@
     on_start/2,
     on_stop/2,
     on_query/3,
-    on_batch_query/3,
+    on_batch_query/4,
     on_get_status/2,
     on_format_query_result/1
 ]).
@@ -197,18 +197,20 @@ on_query(
 on_batch_query(
     InstId,
     BatchReq = [{Key, _} | _],
-    #{query_templates := Templates} = State
+    #{query_templates := Templates} = State,
+    ChannelConfig
 ) ->
     case maps:get({Key, batch}, Templates, undefined) of
         undefined ->
             {error, {unrecoverable_error, batch_select_not_implemented}};
         Template ->
-            on_batch_insert(InstId, BatchReq, Template, State)
+            on_batch_insert(InstId, BatchReq, Template, State, ChannelConfig)
     end;
 on_batch_query(
     InstId,
     BatchReq,
-    State
+    State,
+    _
 ) ->
     ?SLOG(error, #{
         msg => "invalid request",
@@ -509,16 +511,20 @@ proc_sql_params(TypeOrKey, SQLOrData, Params, #{query_templates := Templates}) -
 proc_sql_params(_TypeOrKey, SQLOrData, Params, _State) ->
     {SQLOrData, Params}.
 
-on_batch_insert(InstId, BatchReqs, {InsertPart, RowTemplate}, State) ->
-    Rows = [render_row(RowTemplate, Msg) || {_, Msg} <- BatchReqs],
+on_batch_insert(InstId, BatchReqs, {InsertPart, RowTemplate}, State, ChannelConfig) ->
+    Rows = [render_row(RowTemplate, Msg, ChannelConfig) || {_, Msg} <- BatchReqs],
     Query = [InsertPart, <<" values ">> | lists:join($,, Rows)],
     on_sql_query(InstId, query, Query, no_params, default_timeout, State).
 
-render_row(RowTemplate, Data) ->
-    % NOTE
-    % Ignoring errors here, missing variables are set to "'undefined'" due to backward
-    % compatibility requirements.
-    RenderOpts = #{escaping => mysql, undefined => <<"undefined">>},
+render_row(RowTemplate, Data, ChannelConfig) ->
+    RenderOpts =
+        case maps:get(undefined_vars_as_null, ChannelConfig, false) of
+            % NOTE:
+            %  Ignoring errors here, missing variables are set to "'undefined'" due to backward
+            %  compatibility requirements.
+            false -> #{escaping => mysql, undefined => <<"undefined">>};
+            true -> #{escaping => mysql}
+        end,
     {Row, _Errors} = emqx_template_sql:render(RowTemplate, {emqx_jsonish, Data}, RenderOpts),
     Row.
 

apps/emqx_rule_engine/src/emqx_rule_funcs.erl

@@ -902,9 +902,9 @@ join_to_sql_values_string(List) ->
         [
             case is_list(Item) of
                 true ->
-                    emqx_placeholder:quote_sql(emqx_utils_json:encode(Item));
+                    emqx_placeholder:quote_sql2(emqx_utils_json:encode(Item));
                 false ->
-                    emqx_placeholder:quote_sql(Item)
+                    emqx_placeholder:quote_sql2(Item)
             end
          || Item <- List
         ],

apps/emqx_utils/src/emqx_placeholder.erl

@@ -20,6 +20,7 @@
 -export([
     preproc_tmpl/1,
     preproc_tmpl/2,
+    proc_nullable_tmpl/2,
     proc_tmpl/2,
     proc_tmpl/3,
     preproc_cmd/1,
@@ -29,22 +30,26 @@
     preproc_sql/2,
     proc_sql/2,
     proc_sql_param_str/2,
+    proc_sql_param_str2/2,
     proc_cql_param_str/2,
     proc_param_str/3,
     preproc_tmpl_deep/1,
     preproc_tmpl_deep/2,
     proc_tmpl_deep/2,
     proc_tmpl_deep/3,
-
     bin/1,
+    nullable_bin/1,
     sql_data/1,
     lookup_var/2
 ]).
 
 -export([
     quote_sql/1,
+    quote_sql2/1,
     quote_cql/1,
-    quote_mysql/1
+    quote_cql2/1,
+    quote_mysql/1,
+    quote_mysql2/1
 ]).
 
 -export_type([tmpl_token/0]).
@@ -111,6 +116,10 @@ preproc_tmpl(Str, Opts) ->
     Tokens = re:split(Str, RE, [{return, binary}, group, trim]),
     do_preproc_tmpl(Opts, Tokens, []).
 
+proc_nullable_tmpl(Tokens, Data) ->
+    Opts = #{return => full_binary, var_trans => fun nullable_bin/1},
+    proc_tmpl(Tokens, Data, Opts).
+
 -spec proc_tmpl(tmpl_token(), map()) -> binary().
 proc_tmpl(Tokens, Data) ->
     proc_tmpl(Tokens, Data, #{return => full_binary}).
@@ -182,6 +191,10 @@ proc_sql_param_str(Tokens, Data) ->
     % https://www.postgresql.org/docs/14/sql-syntax-lexical.html#SQL-SYNTAX-CONSTANTS
     proc_param_str(Tokens, Data, fun quote_sql/1).
 
+-spec proc_sql_param_str2(tmpl_token(), map()) -> binary().
+proc_sql_param_str2(Tokens, Data) ->
+    proc_param_str(Tokens, Data, fun quote_sql2/1).
+
 -spec proc_cql_param_str(tmpl_token(), map()) -> binary().
 proc_cql_param_str(Tokens, Data) ->
     proc_param_str(Tokens, Data, fun quote_cql/1).
@@ -238,6 +251,10 @@ proc_tmpl_deep({tmpl, Tokens}, Data, Opts) ->
 proc_tmpl_deep({tuple, Elements}, Data, Opts) ->
     list_to_tuple([proc_tmpl_deep(El, Data, Opts) || El <- Elements]).
 
+nullable_bin(undefined) -> <<"null">>;
+nullable_bin(null) -> <<"null">>;
+nullable_bin(Var) -> bin(Var).
+
 -spec sql_data(term()) -> term().
 sql_data(undefined) -> null;
 sql_data(List) when is_list(List) -> List;
@@ -254,14 +271,26 @@ bin(Val) -> emqx_utils_conv:bin(Val).
 quote_sql(Str) ->
     emqx_utils_sql:to_sql_string(Str, #{escaping => sql, undefined => <<"undefined">>}).
 
+-spec quote_sql2(_Value) -> iolist().
+quote_sql2(Str) ->
+    emqx_utils_sql:to_sql_string(Str, #{escaping => sql}).
+
 -spec quote_cql(_Value) -> iolist().
 quote_cql(Str) ->
     emqx_utils_sql:to_sql_string(Str, #{escaping => cql, undefined => <<"undefined">>}).
 
+-spec quote_cql2(_Value) -> iolist().
+quote_cql2(Str) ->
+    emqx_utils_sql:to_sql_string(Str, #{escaping => cql}).
+
 -spec quote_mysql(_Value) -> iolist().
 quote_mysql(Str) ->
     emqx_utils_sql:to_sql_string(Str, #{escaping => mysql, undefined => <<"undefined">>}).
 
+-spec quote_mysql2(_Value) -> iolist().
+quote_mysql2(Str) ->
+    emqx_utils_sql:to_sql_string(Str, #{escaping => mysql}).
+
 lookup_var(Var, Value) when Var == ?PH_VAR_THIS orelse Var == [] ->
     Value;
 lookup_var([Prop | Rest], Data0) ->

changes/ee/feat-13861.en.md

@@ -0,0 +1,10 @@
+A new configuration item `undefined_vars_as_null` has been added to some of the databases actions, to ensure that undefined variables are treated as NULL when writing data.
+
+The following actions are affected by this configuration item:
+
+Actions:
+- MySQL
+- ClickHouse
+- SQLServer
+- TDengine
+- DynamoDB

rel/i18n/emqx_bridge_v2_schema.hocon

@@ -22,4 +22,12 @@ config_enable.desc:
 config_enable.label:
 """Enable or Disable"""
 
+undefined_vars_as_null.desc:
+"""When writing to databases, treat undefined variables as NULL.
+When this option is enabled, if undefined variables (like ${var}) are used in templates, they will be replaced with "NULL" instead of the string "undefined". If this option is not enabled (default), the string "undefined" might be inserted.
+This option should always be `true` if possible; the default value `false` is only to ensure backward compatibility."""
+
+undefined_vars_as_null.label:
+"""Undefined Vars as Null"""
+
 }