yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(tdengine): fix SQL template errors 

1. supports multi-table insert
2. supports constructing subtable names by mixed prefixes and placeholders
This commit is contained in:
firest 2023-07-13 04:44:48 +00:00
parent 8ebe099653
commit fef5087c41
1 changed files with 11 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
apps/emqx_bridge_tdengine/src/emqx_bridge_tdengine_connector.erl
View File

@ -124,7 +124,7 @@ on_query(InstanceId, {query, SQL}, State) ->
on_query(InstanceId, {Key, Data}, #{insert_tokens := InsertTksMap} = State) -> on_query(InstanceId, {Key, Data}, #{insert_tokens := InsertTksMap} = State) ->
case maps:find(Key, InsertTksMap) of case maps:find(Key, InsertTksMap) of
{ok, Tokens} when is_map(Data) -> {ok, Tokens} when is_map(Data) ->
SQL = emqx_placeholder:proc_sql_param_str(Tokens, Data), SQL = emqx_placeholder:proc_tmpl(Tokens, Data),
do_query(InstanceId, SQL, State); do_query(InstanceId, SQL, State);
_ -> _ ->
{error, {unrecoverable_error, invalid_request}} {error, {unrecoverable_error, invalid_request}}
@ -209,31 +209,16 @@ execute(Conn, Query, Opts) ->
tdengine:insert(Conn, Query, Opts). tdengine:insert(Conn, Query, Opts).
do_batch_insert(Conn, Tokens, BatchReqs, Opts) -> do_batch_insert(Conn, Tokens, BatchReqs, Opts) ->
Queries = aggregate_query(Tokens, BatchReqs), SQL = aggregate_query(Tokens, BatchReqs, <<"INSERT INTO">>),
SQL = maps:fold(
fun(InsertPart, Values, Acc) ->
lists:foldl(
fun(ValuePart, IAcc) ->
<<IAcc/binary, " ", ValuePart/binary>>
end,
<<Acc/binary, " ", InsertPart/binary, " VALUES">>,
Values
)
end,
<<"INSERT INTO">>,
Queries
),
execute(Conn, SQL, Opts). execute(Conn, SQL, Opts).
aggregate_query({InsertPartTks, ParamsPartTks}, BatchReqs) -> aggregate_query(BatchTks, BatchReqs, Acc) ->
lists:foldl( lists:foldl(
fun({_, Data}, Acc) -> fun({_, Data}, InAcc) ->
InsertPart = emqx_placeholder:proc_sql_param_str(InsertPartTks, Data), InsertPart = emqx_placeholder:proc_tmpl(BatchTks, Data),
ParamsPart = emqx_placeholder:proc_sql_param_str(ParamsPartTks, Data), <<InAcc/binary, " ", InsertPart/binary>>
Values = maps:get(InsertPart, Acc, []),
maps:put(InsertPart, [ParamsPart | Values], Acc)
end, end,
#{}, Acc,
BatchReqs BatchReqs
). ).
@ -260,13 +245,12 @@ parse_batch_prepare_sql([{Key, H} | T], InsertTksMap, BatchTksMap) ->
InsertTks = emqx_placeholder:preproc_tmpl(H), InsertTks = emqx_placeholder:preproc_tmpl(H),
H1 = string:trim(H, trailing, ";"), H1 = string:trim(H, trailing, ";"),
case split_insert_sql(H1) of case split_insert_sql(H1) of
[_InsertStr, InsertPart, _ValuesStr, ParamsPart] -> [_InsertPart, BatchDesc] ->
InsertPartTks = emqx_placeholder:preproc_tmpl(InsertPart), BatchTks = emqx_placeholder:preproc_tmpl(BatchDesc),
ParamsPartTks = emqx_placeholder:preproc_tmpl(ParamsPart),
parse_batch_prepare_sql( parse_batch_prepare_sql(
T, T,
InsertTksMap#{Key => InsertTks}, InsertTksMap#{Key => InsertTks},
BatchTksMap#{Key => {InsertPartTks, ParamsPartTks}} BatchTksMap#{Key => BatchTks}
); );
Result -> Result ->
?SLOG(error, #{msg => "split sql failed", sql => H, result => Result}), ?SLOG(error, #{msg => "split sql failed", sql => H, result => Result}),
@ -299,7 +283,7 @@ split_insert_sql(SQL0) ->
{true, E1} {true, E1}
end end
end, end,
re:split(SQL, "(?i)(insert into)|(?i)(values)") re:split(SQL, "(?i)(insert into)")
). ).
formalize_sql(Input) -> formalize_sql(Input) ->