%%% vim:ts=2:sw=2:et
%%%-----------------------------------------------------------------------------
%%% @doc CSV file parsing functions
%%%
%%% @author Serge Aleynikov <saleyn@gmail.com>
%%% @copyright 2021 Serge Aleynikov
%%% @end
%%%-----------------------------------------------------------------------------
%%% Created 2021-06-01
%%%-----------------------------------------------------------------------------
-module(csv).
-author('saleyn@gmail.com').
-export([parse/1, parse/2, parse_line/1]).
-export([max_field_lengths/2, guess_data_types/2, guess_data_type/1, load_to_mysql/4]).
-type load_options() ::
[{load_type, recreate|replace|ignore_dups|update_dups}|
{col_types, #{binary() => ColType::atom() | {ColType::atom(), ColLen::integer()}}}|
{batch_size, integer()}|
{blob_size, integer()}|
{create_table, boolean()}|
{save_create_sql_to_file, string()}|
{transforms, #{binary() => fun((term()) -> term())}}|
{guess_types, boolean()}|
{guess_limit_rows, integer()}|
{max_nulls_pcnt, float()}|
{primary_key, PKColumns::binary()|[binary()|list()]}|
{drop_temp_table, boolean()}|
{encoding, string()|atom()}|
{verbose, boolean()|integer()}].
%% Options for loading data to a database.
%% <dl>
%% <dt>{load_type, Type}</dt>
%% <dd>Type of loading to perform. `recreate' will replace the table by atomically
%% dropping the old one, creating/loading the new one, and replacing the table.
%% `replace' will do an insert by using `REPLACE INTO' statement. `ignore_dups'
%% will use `INSERT IGNORE INTO' statement to ignore records with duplicate keys.
%% `update_dups' will do an `INSERT INTO' and `ON DUPLICATE KEY UPDATE', so that
%% the old records are updated and the new ones are inserted.</dd>
%% <dt>{create_table, Allow}</dt>
%% <dd>Allow to create a table if it doesn't exist</dd>
%% <dt>{col_types, Map}</dt>
%% <dd>Types of data for all or some columens. The Map is in the format:
%% `ColName::binary() => ColInfo', where
%% ColInfo is `ColType | {ColType, ColLen::integer()}', and `ColType' is:
%% `date | datetime | integer | float | blob | number'.
%% </dd>
%% <dt>{transforms, Map}</dt>
%% <dd>Value transformation function for columns. The Map is in the format:
%% `ColName::binary() => fun((Value::term()) -> term())'.</dd>
%% <dt>{batch_size, Size}</dt>
%% <dd>Number of records per SQL insert/update/replace call</dd>
%% <dt>{blob_size, Size}</dt>
%% <dd>Threshold in number of bytes at which a VARCHAR field is defined as BLOB</dd>
%% <dt>{save_create_sql_to_file, Filename::string()}</dt>
%% <dd>Save CREATE TABLE sql statement to a file</dd>
%% <dt>guess_types</dt>
%% <dd>When specified, the function will try to guess the type of data in columns
%% instead of treating all data as string fields. The possible data typed guessed:
%% integer, float, date, datetime, number, string</dd>
%% <dt>{guess_limit_rows, Limit}</dt>
%% <dd>Limit the number of rows for guessing the column data types</dd>
%% <dt>{max_nulls_pcnt, Percent}</dt>
%% <dd>A percentage threshold of permissible NULLs in a column (0-100), above which
%% the column data type is forced to be treated as `string'</dd>
%% <dt>{primary_key, Fields}</dt>
%% <dd>Names of primary key fields in the created table</dd>
%% <dt>{drop_temp_table, boolean()}</dt>
%% <dd>When true (default), temp table is dropped.</dd>
%% <dt>{encoding, Encoding}</dt>
%% <dd>The name of the encoding to use for storing data. For the list of permissible
%% values [see this
%% link](https://dev.mysql.com/doc/refman/8.0/en/charset-unicode-sets.html)</dd>
%% <dt>verbose</dt>
%% <dd>Print additional details to stdout</dd>
%% </dl>
-export_type([load_options/0]).
-type parse_options() ::
[fix_lengths |
binary |
list |
{open, list()} |
{columns, [binary() |string()]}|
{converters, [{binary()|string()|all, fun((binary(), binary()) -> binary())|
{rex, binary(), binary()}}]}
].
%% CSV Parsing Options.
%% <dl>
%% <dt>fix_lengths</dt><dd>if a record has a column count
%% greater than what's found in the header row, those extra columns will be
%% dropped, and if a row has fewer columns, empty columns will be added.</dd>
%% <dt>binary</dt><dd>Return fields as binaries (default)</dd>
%% <dt>list</dt><dd>Return fields as lists</dd>
%% <dt>{open, list()}</dt><dd>Options given to file:open/2</dd>
%% <dt>{columns, Names}</dt><dd>Return data only in given columns</dd>
%% <dt>{converters, [{Col, fun((ColName, Value) -> NewValue)|
%% {rex, RegEx, Replace}]}</dt>
%% <dd>Data format converter.
%% If `Col' is `all', the same converting function is used for all
%% columns. If the converter is a `{rex, RegEx, Replace}' then
%% the regular expression replacement will be run on a value in the
%% requested column.</dd>
%% </dl>
-export_type([parse_options/0]).
%%------------------------------------------------------------------------------
%% CSV parsing
%%------------------------------------------------------------------------------
%% @doc Parse a CSV file using default options.
-spec parse(string()) -> [[binary()]].
parse(File) when is_list(File); is_binary(File) ->
parse(File, []).
%%------------------------------------------------------------------------------
%% @doc Parse a given CSV file.
%% @end
%%------------------------------------------------------------------------------
-spec parse(binary()|string(), parse_options()) -> [[string()]].
parse(File, Opts) when is_binary(File) ->
parse(binary_to_list(File), Opts);
parse(File, Opts) when is_list(File), is_list(Opts) ->
FileOpts = proplists:get_value(open, Opts, []),
Mode = case proplists:get_value(binary, Opts, true) andalso
not proplists:get_value(list, Opts, false)
of
true -> binary;
false -> list
end,
Columns = case proplists:get_value(columns, Opts, all) of
all -> all;
L0 -> [to_binary(I) || I <- L0]
end,
{ok, F} = file:open(File, [read, raw, binary]++FileOpts),
FstLine = case file:read_line(F) of
{ok, <<I/utf8, Line/binary>>} when I == 65279 -> %% Utf-8 <<239,187,191,...>>
{ok, Line};
Other ->
Other
end,
FixLen = proplists:get_value(fix_lengths, Opts, false),
CSV = case parse_csv_file(F, 1, FstLine, []) of
[] ->
[];
LL when FixLen ->
HLen = length(hd(LL)),
[fix_length(HLen, length(R), R) || R <- LL];
LL ->
LL
end,
% Get converters, which are in the form {ColName|all, fun((V) -> NewV)}.
Converts =
case {CSV, proplists:get_value(converters, Opts, none)} of
{[], _} ->
[];
{[Headers|_], none} ->
list_to_tuple(lists:duplicate(length(Headers), undefined));
{[Headers|_], L1} when is_list(L1) ->
L2 = [{if H==all -> all; true -> to_binary(H) end,
if is_function(Fun,2) ->
Fun;
element(1,Fun)==rex ->
Fun;
true ->
throw("Invalid converter type for column: " ++ to_string(H))
end} || {H,Fun} <- L1],
list_to_tuple([case proplists:get_value(I, L2, proplists:get_value(all, L2)) of
Fun when is_function(Fun,2); element(1,Fun)==rex ->
Fun;
undefined ->
undefined
end || I <- Headers])
end,
case CSV of
[] ->
[];
[HD|Rows] ->
BitMask = list_to_tuple([Columns==all orelse lists:member(I, Columns) || I <- HD]),
%% Get resulting column headers
H = foldr(fun(I, Col, S) ->
case element(I, BitMask) of
true -> [if Mode==list -> binary_to_list(Col); true -> Col end | S];
false -> S
end
end, [], HD),
%% Get resulting data rows
{D,_} = lists:mapfoldl(
fun(Row, RowNum) ->
NewRow = foldr(fun(I, V, A) ->
case element(I,BitMask) of
true -> [convert(Converts, RowNum, I, V, Mode)|A];
false -> A
end
end, [], Row),
{NewRow, RowNum+1}
end,
1,
Rows),
[H | D]
end.
convert(CvtMask, RowNum, I, V, Mode) ->
case element(I, CvtMask) of
undefined ->
if Mode == list -> binary_to_list(V);
true -> V
end;
Fun when is_function(Fun, 2) ->
if Mode == list ->
binary_to_list(Fun(V, RowNum));
true ->
Fun(V, RowNum)
end;
{rex, RegEx, Replace} ->
re:replace(V, RegEx, Replace, [{return,Mode}]);
{rex, RegEx, Replace, Options} ->
re:replace(V, RegEx, Replace, [{return,Mode}|Options])
end.
parse_csv_file(F, _, eof, Done) ->
file:close(F),
lists:reverse(Done);
parse_csv_file(F, LineNo, {ok, Line}, Done) ->
parse_csv_file(F, LineNo+1, file:read_line(F), [parse_line(LineNo, Line)|Done]);
parse_csv_file(_F, LineNo, {error, Reason}, _) ->
throw({error, [{line, LineNo}, {reason, file:format_error(Reason)}]}).
trim_eol(Line) ->
trim_eol(0, Line).
trim_eol(N, Line) when N < byte_size(Line) ->
Z = byte_size(Line),
M = N+1,
case binary:at(Line, Z-M) of
C when C == $\r; C == $\n; C == $ ->
trim_eol(M, Line);
_ ->
Z-N
end;
trim_eol(N, Line) ->
byte_size(Line) - N.
%% @doc Parse a CSV line
-spec parse_line(binary()) -> list().
parse_line(Line) ->
parse_line(1, Line).
parse_line(LineNo, Line) ->
try
parse_csv_field(0, trim_eol(Line), Line, 0,0, [], false)
catch E:R:S ->
erlang:raise(E, {line_parse_error, LineNo, Line, R}, S)
end.
field(Line, Pos, Len, true = _HasEscDblQuote) ->
binary:replace(binary:part(Line, Pos, Len), <<"\"\"">>, <<"\"">>, [global]);
field(Line, Pos, Len, _) ->
binary:part(Line, Pos, Len).
trim(_Line, I, I, _Inc) ->
I;
trim(Line, I, End, Inc) ->
case binary:at(Line, I) of
$ ->
trim(Line, I+Inc, End, Inc);
_ ->
I
end.
trim(Line, Pos, End) when Pos < End ->
Front = trim(Line, Pos, End, 1),
Back = trim(Line, End, Front, -1),
{Front, Back+1};
trim(_Line, Pos, End) ->
{Pos, End+1}.
parse_csv_field(From, To, Line, Pos,Len, Fields, HasEscDblQuote) when From >= To ->
{Start, End} = trim(Line, Pos, Pos+Len-1),
lists:reverse([field(Line, Start, End-Start, HasEscDblQuote) | Fields]);
parse_csv_field(From, To, Line, Pos,Len, Fields, HasEscDblQuote) ->
case Line of
<<_:From/binary, "\"", _/binary>> ->
parse_csv_field_q(From+1, To, Line, From+1, 0, Fields, HasEscDblQuote);
<<_:From/binary, ",", _/binary>> ->
{Start, End} = trim(Line, Pos, Pos+Len-1),
parse_csv_field(From+1,To,Line, From+1, 0, [field(Line,Start,End-Start,HasEscDblQuote)|Fields], false);
_ ->
parse_csv_field(From+1,To,Line, Pos,Len+1, Fields, HasEscDblQuote)
end.
parse_csv_field_q(From, To, Line, Pos,Len, Fields, HasEscDblQuote) when From < To ->
case Line of
<<_:From/binary, "\"\"", _/binary>> ->
parse_csv_field_q(From+2, To, Line, Pos, Len+2, Fields, true);
<<_:From/binary, "\\", _/binary>> ->
parse_csv_field_q(From+1, To, Line, Pos, Len+1, Fields, HasEscDblQuote);
<<_:From/binary, "\"", _/binary>> ->
parse_csv_field (From+1, To, Line, Pos, Len, Fields, HasEscDblQuote);
_ ->
parse_csv_field_q(From+1, To, Line, Pos, Len+1, Fields, HasEscDblQuote)
end;
parse_csv_field_q(From, To, Line, Pos,Len, Fields, HasEscDblQuote) ->
parse_csv_field(From, To, Line, Pos,Len, Fields, HasEscDblQuote).
fix_length(HLen, HLen, Data) -> Data;
fix_length(HLen, DLen, Data) when HLen < DLen ->
{RR, _} = lists:split(HLen, Data),
RR;
fix_length(HLen, DLen, Data) when HLen > DLen ->
Value = if is_binary(hd(Data)) -> <<"">>; true -> "" end,
RR = lists:duplicate(HLen-DLen, Value),
Data ++ RR.
%%------------------------------------------------------------------------------
%% @doc Get max field lengths for a list obtained by parsing a CSV file with
%% `parse_csv_file(File,[fix_lengths])'.
%% @end
%%------------------------------------------------------------------------------
-spec max_field_lengths(HasHeaderRow::boolean(), Rows::[Fields::list()]) -> [Len::integer()].
max_field_lengths(true = _HasHeaders, [_Headers | Rows ] = _CSV) ->
max_field_lengths(false, Rows);
max_field_lengths(false = _HasHeaders, CsvRows) ->
% Calculate length of a field in a row:
% CSV -> [[R1FieldLen0, ..., R1FieldLenN], ...,
% [RnFieldLen0, RnFieldLenN]]
F = fun(Row) -> [if is_binary(I) -> byte_size(I); true -> length(I) end || I <- Row] end,
CsvL = [F(Row) || Row <- CsvRows],
MLens =
fun
G([[] | _], Acc) ->
lists:reverse(Acc);
G(RRows, Acc) ->
{MaxLen, F1Nrows} = lists:foldl(fun([H|T], {A,L}) -> {max(H,A),[T|L]} end, {0,[]}, RRows),
G(F1Nrows, [MaxLen | Acc])
end,
MLens(CsvL, []).
%%------------------------------------------------------------------------------
%% @doc Guess data types of fields in the given CSV list of rows obtained by
%% parsing a CSV file with `parse(File,[fix_lengths])'.
%% The function returns a list of tuples `{Type, MaxFieldLen, NumOfNulls}',
%% where the `Type' is a field type, `MaxFieldLen' is the max length of
%% data in this column, and `NumOfNulls' is the number of rows with empty
%% values in this column.
%% @end
%%------------------------------------------------------------------------------
-spec guess_data_types(HasHeaderRow::boolean(), Rows::[Fields::[binary()]]) ->
{Type::string|integer|number|float|date|datetime,
MaxFieldLen::integer(),
NumOfNulls::integer()}.
guess_data_types(HasHeaders, CSV) when is_boolean(HasHeaders), is_list(CSV) ->
guess_data_types(HasHeaders, CSV, 20.0, 1_000_000).
-spec guess_data_types(HasHeaderRow::boolean(), Rows::[Fields::list()], number(), integer()) ->
{[{FieldType::string|date|datetime|integer|float|number,
MaxFieldLength::integer(),
NullsCount::integer()}],
[Row::[term()]]}.
guess_data_types(true = _HasHeaders, [Headers | Rows ] = _CSV, NullsMaxPcnt, SniffRows) ->
{TpLenNulls, DataRows} = guess_data_types(false, Rows, NullsMaxPcnt, SniffRows),
{TpLenNulls, [Headers | DataRows]};
guess_data_types(false = _HasHeaders, CsvRows, NullsMaxPcnt, SniffRows)
when is_number(NullsMaxPcnt), NullsMaxPcnt >= 0.0, NullsMaxPcnt =< 100.0
, is_integer(SniffRows) ->
% CSV -> [[R1Field0, ..., R1FieldN], ..., [RnField0, RnFieldN]]
NRows = length(CsvRows),
F = fun(UseString, Row) ->
[guess_data_type2(I,UseString) || I <- Row]
end,
CsvL = [F(I > SniffRows, Row) || {I, Row} <- lists:zip(lists:seq(1,NRows), CsvRows)],
Res = scan_column(CsvL, NullsMaxPcnt, NRows),
Data = [ [V || {_,V,_} <- R] || R <- CsvL],
{Res, Data}.
sort_fun(A,A) -> true;
sort_fun(null,_) -> true;
sort_fun(_,null) -> false;
sort_fun(A,B) -> A =< B.
scan_column([], _NullsMaxPcnt, _NRows) ->
[];
scan_column(L, NullsMaxPcnt, NRows) when is_number(NullsMaxPcnt) ->
scan_column(L, NRows, NullsMaxPcnt, []).
scan_column([[]|_], _NRows, _NullsMaxPcnt, Acc) ->
lists:reverse(Acc);
scan_column(L, NRows, NullsMaxPcnt, Acc) ->
{_Type, _MLen, _Nulls} = Res = scan_mlt(L, NRows, NullsMaxPcnt),
Acc1 = [Res | Acc],
case L of
[[_|_]|_] ->
scan_column([tl(I) || I <- L], NRows, NullsMaxPcnt, Acc1);
_ ->
scan_column([[]], NRows, NullsMaxPcnt, Acc1)
end.
scan_mlt(L, NRows, NullsMaxPcnt) when is_number(NullsMaxPcnt) ->
{CTypes, MLen, Nulls, _} = scan_mlt2(L, {#{}, 0, 0, false}),
PcntNulls = if NRows == 0 -> 100; true -> Nulls / NRows * 100 end,
Fun = fun
([T]) when T /= null -> T;
([date,datetime]) -> datetime;
([float,integer]) -> number;
(_) -> string
end,
Type =
case lists:sort(fun sort_fun/2, maps:keys(CTypes)) of
[null|T] when PcntNulls =< NullsMaxPcnt ->
Fun(T);
T ->
Fun(T)
end,
{Type, MLen, Nulls}.
scan_mlt2([[{I,_,V} | _] | T], {ATypes, MLen, Nulls, HasStrings}) ->
case I of
null ->
scan_mlt2(T, {ATypes#{null => 1}, MLen, Nulls+1, HasStrings});
string when HasStrings ->
scan_mlt2(T, {ATypes, max(byte_size(V),MLen), Nulls, true});
string ->
scan_mlt2(T, {ATypes#{string => 1}, max(byte_size(V),MLen), Nulls, true});
Type ->
scan_mlt2(T, {ATypes#{Type => 1}, max(byte_size(V),MLen), Nulls, HasStrings})
end;
scan_mlt2([[]|_], Acc) ->
Acc;
scan_mlt2([], Acc) ->
Acc.
%%------------------------------------------------------------------------------
%% @doc Load CSV data from a `File' to a `MySQL' database.
%% `Tab' is the name of a table where to load data. `MySqlPid' is the pid of a
%% MySQL database connection returned by `mysql:start_link/1'.
%% The data in the table is replaced according to `{import_type, Type}':
%% <ul>
%% <li>`recreate' - The table is entirely replaced with the data from file.
%% The data from the file is loaded atomically - i.e. either
%% the whole file loading succeeds or fails. This is accomplished by first
%% loading data to a temporary table, and then using the database's ACID
%% properties to replace the target table with the temporary table.</li>
%% <li>`replace' - Use "REPLACE INTO" instead of "INSERT INTO" existing
%% table</li>
%% <li>`ignore_dups' - The insert in the existing table is performed and the
%% records with duplicate keys are ignored</li>
%% <li>`update_dups' - The insert in the existing table is performed and the
%% records with duplicate keys are updated</li>
%% <li>`upsert' - The insert/update in the existing table is performed
%% without creating a temporary table</li>
%% </ul>
%%
%% NOTE: this function requires [https://github.com/mysql-otp/mysql-otp.git]
%% @end
%%------------------------------------------------------------------------------
-spec load_to_mysql(File::string(), Tab::string(),
MySqlPid::pid(), Opts::load_options()) ->
{Columns::list(), AffectedCount::integer(), RecCount::integer()}.
load_to_mysql(File, Tab, MySqlPid, Opts)
when is_list(File), is_list(Tab), is_pid(MySqlPid), is_list(Opts) ->
try
BatSz = proplists:get_value(batch_size, Opts, 100), % Insert this many records per call
BlobSz = proplists:get_value(blob_size, Opts, 1000),% Length at which VARCHAR becomes BLOB
Create = proplists:get_value(create_table,Opts, true),% Allow to create table
Types = proplists:get_value(col_types, Opts, #{}), % Column types
Trans0 = proplists:get_value(transforms, Opts, #{}), % Column transforms
Enc = encoding(proplists:get_value(encoding, Opts, undefined)),
Verbose= case proplists:get_value(verbose, Opts, 0) of
true -> 1;
false -> 0;
_I when is_integer(_I) -> _I
end,
SaveSQL= proplists:get_value(save_create_sql_to_file, Opts, false),
LoadTp = proplists:get_value(load_type, Opts, recreate),
lists:member(LoadTp,[recreate, replace, ignore_dups, update_dups, upsert])
orelse throw({badarg, {load_type, LoadTp}}),
Drop = proplists:get_value(drop_temp_table, Opts, true),
PKey = case proplists:get_value(primary_key, Opts, undefined) of
K when is_binary(K) ->
[K];
[H|_] = K when is_binary(H); is_list(H) ->
[to_binary(I) || I <- K];
undefined ->
[];
[] ->
[];
Other ->
throw({badarg, {primary_key, Other}})
end,
CSV0 = parse(File, [fix_lengths, binary]),
ColCnt = length(hd(CSV0)),
CSV1 = [[list_to_binary(cleanup_header(to_string(S))) || S <- hd(CSV0)] | tl(CSV0)],
Heads = hd(CSV1),
%% If no records to insert, then just short-circuit the processing
tl(CSV1) == [] andalso
throw({ignore, {Heads, 0, 0}}),
PKey /= [] andalso
lists:foreach(fun(K) ->
lists:member(K, Heads) orelse throw({primary_key_not_found, K, Heads})
end, PKey),
Merge = fun M([], []) -> [];
M([H|T1], [{T,I,J}|T2]) -> [{H,T,I,J}|M(T1,T2)];
M([H|T1], [I|T2]) when is_integer(I) -> [{H,string,I,0}|M(T1,T2)]
end,
{ColNmTpMxLens, CSV} =
case proplists:get_value(guess_types, Opts, false) of
true ->
SniffRows = proplists:get_value(guess_limit_rows, Opts, 1_000_000),
MaxNullsPcnt = proplists:get_value(max_nulls_pcnt, Opts, 40.0),
{TLN, CSV2} = guess_data_types(true, CSV1, MaxNullsPcnt, SniffRows),
{Merge(Heads, TLN), CSV2};
false ->
{Merge(Heads, max_field_lengths(true, CSV1)), CSV1}
end,
Verbose > 0 andalso
io:format(standard_error,
"LoadType: ~p\n"
"Columns:\n ~p\n", [LoadTp, ColNmTpMxLens]),
%% Check if the table exists
Exists = case mysql:query(MySqlPid, "SELECT count(*) as a\n"
" FROM information_schema.tables\n"
" WHERE table_schema=database() AND\n"
" table_name='" ++ Tab ++ "';\n")
of
{ok,_,[[N]]} ->
N > 0;
{error, {Code01, _, Msg01}} ->
throw({error_checking_if_table_exists, Code01, binary_to_list(Msg01), Tab})
end,
TmpTab = case LoadTp of
upsert -> Tab;
_ -> Tab ++ "_tmp"
end,
CrTab = lists:flatten([
if LoadTp == upsert -> [];
true -> ["DROP TABLE IF EXISTS `", TmpTab, "`;\n"]
end,
case Exists of
true when LoadTp==replace
; LoadTp==update_dups
; LoadTp==ignore_dups
;(LoadTp==recreate andalso not Create) ->
["CREATE TABLE `", TmpTab, "` AS SELECT * FROM `", Tab, "` where false;\n"];
true when LoadTp /= upsert
, LoadTp /= recreate ->
throw({invalid_load_type, LoadTp});
false when not Create
, LoadTp /= recreate ->
throw("Table "++Tab++" doesn't exist and creation not allowed!");
_ ->
["CREATE TABLE `", TmpTab, "` (",
string:join(
[
begin
{CTp, I} =
case maps:get(S, Types, T) of
CA when is_atom(CA) ->
{CA, CLen};
{CA, CLen1} ->
{CA, CLen1}
end,
case CTp of
blob -> io_lib:format("`~s` BLOB", [to_string(S)]);
_ when I > BlobSz -> io_lib:format("`~s` BLOB", [to_string(S)]);
date -> io_lib:format("`~s` DATE", [to_string(S)]);
datetime -> io_lib:format("`~s` DATETIME", [to_string(S)]);
integer -> io_lib:format("`~s` BIGINT", [to_string(S)]);
float -> io_lib:format("`~s` DOUBLE", [to_string(S)]);
number -> io_lib:format("`~s` DOUBLE", [to_string(S)]);
_ -> io_lib:format("`~s` VARCHAR(~w)",[to_string(S),I])
end
end || {S,T,CLen,_} <- ColNmTpMxLens], ","),
");\n"]
end,
%% Add primary key
case PKey of
[] -> "";
_ -> SS = string:join(["`"++binary_to_list(S)++"`" || S <- PKey], ","),
io_lib:format("ALTER TABLE `~s` ADD PRIMARY KEY (~s);\n", [TmpTab, SS])
end]),
case SaveSQL of
false ->
ok;
true ->
SQLF = filename:join(filename:dirname(File), filename:basename(File, ".csv") ++ ".sql"),
ok = file:write_file(SQLF, CrTab);
SQLF when is_list(SQLF) ->
ok = file:write_file(SQLF, CrTab)
end,
if CrTab == [] ->
ok;
true ->
Verbose > 0 andalso io:format(standard_error, "SQL:\n====\n~s\n", [CrTab]),
case mysql:query(MySqlPid, CrTab) of
ok -> ok;
{error, {Code, _, Msg}} ->
throw({error_creating_table, Code, binary_to_list(Msg), CrTab})
end
end,
%% Remove NOT NULL constraints from all columnts
if Exists, LoadTp == update_dups ->
{ok, _, NullCols} =
mysql:query(MySqlPid,
"select column_name,column_type,collation_name,column_default,extra "
"from information_schema.columns "
"where table_schema=database() AND table_name='"++Tab++"' "
"and IS_NULLABLE='NO' and COLUMN_KEY!='PRI';"),
NotNullSQL = lists:flatten(
[["ALTER TABLE `",TmpTab,"` MODIFY `",cc(C),"` ",cc(Tp)," NULL",
cc("COLLATE",CN),cc("DEFAULT",CD),cc(EXT),";\n"]
|| [C,Tp,CN,CD,EXT] <- NullCols]),
Verbose > 0 andalso io:format(standard_error, "Replacing NULL columns:\n ~p\n", [NotNullSQL]),
ok = mysql:query(MySqlPid, NotNullSQL);
true ->
ok
end,
[HD|Rows] = CSV,
Trans = lists:foldl(fun(H,{I,A}) ->
{I+1, setelement(I, A, maps:get(H, Trans0, undefined))}
end, {1, list_to_tuple(lists:duplicate(length(HD), undefined))}, HD),
Transform =
fun
G([V|Rest], I) ->
case element(I, Trans) of
F when is_function(1, F) -> [check_null(F(V)) | G(Rest, I+1)];
undefined -> [check_null(V) | G(Rest, I+1)]
end;
G([], _) ->
[]
end,
[_|QQQ0s] = string:copies(",?", ColCnt),
HHHs = string:join(["`"++binary_to_list(S)++"`" || S <- HD], ","),
QQQs = lists:append([",(", QQQ0s, ")"]),
ModRows = case Trans0 of
#{} -> [[check_null(B) || B <- Row] || Row <- Rows];
_ -> [Transform(Row, 1) || Row <- Rows]
end,
BatchRows = stringx:batch_split(BatSz, ModRows),
FstBatLen = length(hd(BatchRows)),
if Enc /= [] ->
Verbose > 0 andalso io:format(standard_error, "SQL:\n====\n~s\n", [Enc]),
ok = mysql:query(MySqlPid, Enc);
true ->
ok
end,
%% Insert data to the temp table
PfxSQL = lists:append(["INSERT INTO `", TmpTab, "` (", HHHs, ") VALUES "]),
[_|SfxSQL]= string:copies(QQQs, FstBatLen),
TailSQL = case LoadTp of
upsert -> on_duplicate_key_update(Tab, HD, PKey);
_ -> []
end,
Verbose > 0 andalso
io:format(standard_error, "SQL:\n====\n~s~s~s\n", [PfxSQL, tl(QQQs), TailSQL]),
{ok,Ref} = mysql:prepare(MySqlPid, PfxSQL++SfxSQL++TailSQL),
NNN = lists:foldl(fun(Batch, I) ->
NRows = length(Batch),
{SqlRef, Unprepare} =
if NRows == FstBatLen ->
{Ref, false};
true ->
[_|Sfx2] = string:copies(QQQs, NRows),
{ok,R} = mysql:prepare(MySqlPid, PfxSQL++Sfx2++TailSQL),
{R, true}
end,
Row = lists:append(Batch),
Verbose > 1 andalso io:format(standard_error, "Inserting: ~p\n", [Row]),
Res = mysql:execute(MySqlPid, SqlRef, Row),
Unprepare andalso mysql:unprepare(MySqlPid, SqlRef),
case Res of
ok ->
I + NRows;
{error, {_Code, _, _Msg}} ->
SQL2 = PfxSQL++tl(QQQs),
try
lists:foldl(fun(R, J) ->
case mysql:query(MySqlPid, SQL2, R) of
ok ->
J+1;
{error, {Code1, _, Msg1}} ->
throw({error_inserting_records, Code1, binary_to_list(Msg1), {row, J, R}})
end
end, I, Batch)
catch _:E ->
mysql:unprepare(MySqlPid, Ref),
throw(E)
end
end
end, 1, BatchRows),
mysql:unprepare(MySqlPid, Ref),
%% Now the temp table is populated - update the actual table
SQL = lists:flatten(
case LoadTp of
upsert ->
[];
recreate ->
OldTab = Tab ++ "_OLD",
%% Rename the actual table X to X_OLD, and the tmp table T to X, and drop X_OLD
%% in one transaction:
["DROP TABLE IF EXISTS `", OldTab, "`;\n"
"-- Check if the real prod table exists\n"
"SELECT count(*) INTO @exists\n"
" FROM information_schema.tables\n"
" WHERE table_schema=database() AND\n"
" table_name='", Tab, "';\n"
"-- If so, atomically rename it to another table,\n"
"-- rename temp table to prod table, and drop the old table\n"
"-- Otherwise, just atomically rename the temp table into prod\n"
"SELECT count(*) from `", TmpTab, "` INTO @cnt;\n"
"SET @query = IF(@exists>0,\n"
" 'RENAME TABLE `", Tab, "` TO `", OldTab, "`, `", TmpTab, "` TO `", Tab, "`',\n"
" 'RENAME TABLE `", TmpTab, "` TO `", Tab, "`');\n"
"PREPARE stmt FROM @query;\n"
"EXECUTE stmt;\n"
"DEALLOCATE PREPARE stmt;\n"
"DROP TABLE IF EXISTS `", OldTab, "`;\n"
"SELECT @cnt,@cnt;\n"];
replace ->
["REPLACE INTO `", Tab, "` (",HHHs,") SELECT ",HHHs," FROM `", TmpTab, "`;\n"
"SELECT ROW_COUNT(),FOUND_ROWS();\n"];
ignore_dups ->
["INSERT IGNORE INTO `", Tab, "` (",HHHs,") SELECT ",HHHs," FROM `", TmpTab, "`;\n"
"SELECT ROW_COUNT(),FOUND_ROWS();\n"];
update_dups ->
["INSERT INTO `", Tab, "` (",HHHs,") SELECT ",HHHs," FROM `", TmpTab, "`\n",
on_duplicate_key_update(Tab, HD, PKey), ";\n",
"SELECT ROW_COUNT(),FOUND_ROWS();\n"]
end),
case SQL of
[] ->
{HD, NNN, NNN};
_ ->
Verbose > 0 andalso io:format(standard_error, "SQL:\n====\n~s\n", [SQL]),
case mysql:query(MySqlPid, SQL) of
{ok, _, [[Changed,SelectedRows]]} ->
if Drop ->
ok = mysql:query(MySqlPid, lists:flatten(["DROP TABLE IF EXISTS `", TmpTab, "`;"]));
true ->
ok
end,
Selected = if SelectedRows < 0 -> 0; true -> SelectedRows end,
{HD, Changed, Selected};
{error, {Code3, _, Msg3}} ->
throw({error_inserting_records, Code3, binary_to_list(Msg3), SQL})
end
end
catch
throw:{ignore, Result} ->
Result
end.
on_duplicate_key_update(Tab, HD, PKey) ->
["ON DUPLICATE KEY UPDATE ",
string:join(
[begin
LL=binary_to_list(S),
["`",LL,"`=IFNULL(VALUES(",LL,"),`",Tab,"`.`",LL,"`)"]
end || S <- HD, not lists:member(S, PKey)],
",")].
encoding(undefined) -> [];
encoding(A) when is_atom(A) -> encoding2(atom_to_list(A));
encoding(L) when is_list(L) -> encoding2(L).
encoding2(L) ->
["SET NAMES ", L, ";\n", encoding3(L)].
encoding3("utf8"++_) ->
"SET CHARACTER SET utf8;\n";
encoding3(Other) ->
["SET CHARACTER SET ", Other].
to_string(L) when is_binary(L) -> binary_to_list(L);
to_string(L) when is_list(L) -> L;
to_string(L) when is_atom(L) -> atom_to_list(L).
to_binary(I) when is_binary(I) -> I;
to_binary(L) when is_list(L) -> list_to_binary(L).
check_null(<<>>) -> null;
check_null(B) -> B.
cc(null) -> "";
cc(B) when is_binary(B) -> binary_to_list(B).
cc(L,null) when is_list(L) -> "";
cc(L,B) when is_binary(B) -> " "++L++" "++binary_to_list(B).
cleanup_header([$ |T]) -> [$_|cleanup_header(T)];
cleanup_header([C|T]) when (C >= $a andalso C =< $z);
(C >= $A andalso C =< $Z);
(C >= $0 andalso C =< $9);
(C == $_)
-> [C|cleanup_header(T)];
cleanup_header([_|T]) -> cleanup_header(T);
cleanup_header([]) -> [].
%% @doc Guess the type of data by its value
-spec guess_data_type(binary()) ->
{null | date | datetime | integer | float | string, term(), string()}.
guess_data_type(S) ->
guess_data_type2(S, false).
guess_data_type2(<<"">>, _) ->
{null, null, <<"">>};
guess_data_type2(S, true) ->
{string, S, S};
guess_data_type2(S, _) ->
guess_data_type3(S).
guess_data_type3(<<C, _/binary>> = S) when C >= $0, C =< $9
; C == $-; C == $+ ->
try
I = binary_to_integer(S),
%case binary_to_integer(S) of
% I when I < -2147483648; I > 2147483647 ->
% {integer, I, S};
% I ->
% {integer, I, S}
%end;
{integer, I, S}
catch _:_ ->
try F = binary_to_float(S), {float, F, S} catch _:_ ->
guess_data_type4(S)
end
end;
guess_data_type3(S) ->
guess_data_type4(S).
guess_data_type4(<<Y1,Y2,Y3,Y4,$-,M1,M2,$-,D1,D2, _/binary>> = V)
when Y1 >= $0, Y1 =< $9, Y2 >= $0, Y2 =< $9, Y3 >= $0, Y3 =< $9, Y4 >= $0, Y4 =< $9
, M1 >= $0, M1 =< $9, M2 >= $0, M2 =< $9
, D1 >= $0, D1 =< $9, D2 >= $0, D2 =< $9
->
Y = i(Y1)*1000 + i(Y2)*100 + i(Y3)*10 + i(Y4),
M = i(M1)*10 + i(M2),
D = i(D1)*10 + i(D2),
if Y < 1000; Y > 2500; M < 1; M > 12; D < 1; D > 31 ->
def_type(V);
true ->
case V of
<<_:10/binary, C,H1,H2,$:,N1,N2,$:,S1,S2, _/binary>>
when (C == $ orelse C == $T)
, H1 >= $0, H1 =< $9, H2 >= $0, H2 =< $9
, N1 >= $0, N1 =< $9, N2 >= $0, N2 =< $9
, S1 >= $0, S1 =< $9, S2 >= $0, S2 =< $9 ->
H = i(H1)*10 + i(H2),
N = i(N1)*10 + i(N2),
S = i(S1)*10 + i(S2),
if H < 0; H > 23; N < 0; N > 59; S < 0; S > 59 ->
def_type(V);
true ->
{datetime, {{Y,M,D},{H,N,S}}, V}
end;
_ when byte_size(V) == 10 ->
{date, {Y,M,D}, V};
_ ->
def_type(V)
end
end;
guess_data_type4(S) ->
def_type(S).
def_type(V) ->
{string, V, V}.
i(C) -> C - $0.
-spec foldr(fun((Position::integer(), Item::term(), Acc::term()) -> NewAcc::term()),
Init::term(), list()) -> term().
foldr(Fun, Init, List) ->
N = length(List),
element(2, lists:foldr(fun(V, {I, S}) -> R = Fun(I, V, S), {I-1, R} end, {N, Init}, List)).
%%------------------------------------------------------------------------------
%% Tests
%%------------------------------------------------------------------------------
-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
parse_test() ->
?assertEqual([<<>>], parse_line(<<"\n">>)),
?assertEqual([<<>>], parse_line(<<"\r\n">>)),
?assertEqual([<<>>], parse_line(<<"\n\r">>)),
?assertEqual([<<>>], parse_line(<<"\n\r\n">>)),
?assertEqual([<<"a">>], parse_line(<<"a\r\n">>)),
?assertEqual([<<>>,<<>>], parse_line(<<",\r\n">>)),
?assertEqual([<<>>,<<>>], parse_line(<<",">>)),
?assertEqual([<<>>,<<"a">>], parse_line(<<",a">>)),
?assertEqual([<<"a">>,<<"b">>], parse_line(<<"a , b">>)),
?assertEqual([<<"a">>,<<"b">>], parse_line(<<" a , b ">>)),
?assertEqual([<<"a">>,<<"b">>], parse_line(<<" a, b ">>)),
?assertEqual([<<"a">>,<<"b">>], parse_line(<<" a,b ">>)),
?assertEqual([<<"a">>,<<"b">>], parse_line(<<" a, \"b\" ">>)),
?assertEqual([<<"a">>,<<"b">>], parse_line(<<" a , b \r\n">>)),
?assertEqual([<<"a">>,<<"b \r\nbb">>,<<"c">>],parse_line(<<"\"a\",\"b \r\nbb\",\"c\"\r\n">>)),
?assertEqual([<<"a\"b">>,<<"c\"">>], parse_line(<<"\"a\"\"b\",\"c\"\"\"">>)),
?assertEqual([<<"a,b">>], parse_line(<<"\"a\,b\"">>)),
?assertEqual([<<"a, b">>], parse_line(<<"\"a\, b\"">>)),
Lines = [<<"a,bb,ccc">>,
<<",b,c">>,
<<",b,\"c,d\"">>,
<<"\"c,d\"\r">>,
<<"xx,yyy,zzzz">>,
<<"xxx,yyyy,zzzzz">>,
<<"a,b\n">>,
<<"x,y\r">>,
<<"z\r\n">>],
CSV = [parse_line(L) || L <- Lines],
Res = [[<<"a">>, <<"bb">>, <<"ccc">>],
[<<"">>, <<"b">>, <<"c">>],
[<<"">>, <<"b">>, <<"c,d">>],
[<<"c,d">>],
[<<"xx">>, <<"yyy">>, <<"zzzz">>],
[<<"xxx">>, <<"yyyy">>, <<"zzzzz">>],
[<<"a">>,<<"b">>],
[<<"x">>,<<"y">>],
[<<"z">>]
],
?assertEqual(Res, CSV).
parse2_test_() ->
Name = case os:type() of
{unix, _} -> "/tmp/csv.test.tmp";
{win32,_} -> os:getenv("TEMP") ++ "/csv.test.tmp"
end,
{setup,
%% Setup
fun() ->
ok = file:write_file(Name, "a,b,c\nr11,r12,r13\nr21,r22,r23\n"),
Name
end,
%% Cleanup
fun(_Name) ->
file:delete(_Name)
end,
%% Tests
[
?_assertEqual([[<<"a">>,<<"b">>,<<"c">>],
[<<"r11">>,<<"r12">>,<<"r13">>],
[<<"r21">>,<<"r22">>,<<"r23">>]
],
parse(Name, [])),
?_assertEqual([[<<"a">>,<<"b">>],
[<<"r11">>,<<"r12">>],
[<<"r21">>,<<"r22">>]
],
parse(Name, [{columns, ["a","b"]}])),
?_assertEqual([[<<"a">>,<<"c">>],
[<<"r11">>,<<"r13">>],
[<<"r21">>,<<"r23">>]
],
parse(Name, [{columns, ["a","c"]}])),
?_assertEqual([[<<"b">>,<<"c">>],
[<<"r12">>,<<"r13">>],
[<<"r22">>,<<"r23">>]
],
parse(Name, [{columns, ["b","c"]}])),
?_assertEqual([["a","b","c"],
["r11","r12","r13"],
["r21","r22","r23"]
],
parse(Name, [list])),
?_assertEqual([["a","b"],
["r11","r12"],
["r21","r22"]
],
parse(Name, [list, {columns, ["a","b"]}])),
%% Test column converters based on a function and a regex:
?_assertEqual([["a","b"],
["XX11","Row1"],
["XX21","Row2"]
],
parse(Name, [list, {columns, ["a","b"]},
{converters,[{"a", {rex, "r(\\d\\d)", "XX\\1"}},
{"b", fun(_, RowNum) ->
<<"Row", (integer_to_binary(RowNum))/binary>>
end}]}]))
]
}.
max_lens_test() ->
Lines = [[<<"a">>, <<"bb">>, <<"ccc">>],
[<<"xx">>, <<"yyy">>, <<"zzzz">>],
[<<"xxx">>, <<"yyyy">>, <<"zzzzz">>]],
Res = max_field_lengths(true, Lines),
?assertEqual([3,4,5], Res).
guess_type_test() ->
?assertEqual({integer, 1, <<"1">>}, guess_data_type(<<"1">>)),
?assertEqual({float, 1.0, <<"1.0">>}, guess_data_type(<<"1.0">>)),
?assertEqual({date, {2021,1,1}, <<"2021-01-01">>}, guess_data_type(<<"2021-01-01">>)),
?assertEqual({datetime,{{2021,1,1},{0,0,0}}, <<"2021-01-01 00:00:00">>}, guess_data_type(<<"2021-01-01 00:00:00">>)),
?assertEqual({datetime,{{2021,1,1},{0,0,0}}, <<"2021-01-01 00:00:00+01:00">>}, guess_data_type(<<"2021-01-01 00:00:00+01:00">>)),
?assertEqual({datetime,{{2021,1,1},{0,0,0}}, <<"2021-01-01 00:00:00-01:00">>}, guess_data_type(<<"2021-01-01 00:00:00-01:00">>)),
?assertEqual({string, <<"abc">>, <<"abc">>}, guess_data_type(<<"abc">>)),
?assertEqual({null, null, <<"">>}, guess_data_type(<<"">>)).
col_types_test() ->
LineS = [["a", "b", "c", "d", "e", "f", "g", "h", "i"],
["1", "1.0", "1.0", "2021-01-02", "2021-01-02", "abc", "1", "1.0","2021-01-02"],
["1", "2.0", "3.0", "2021-03-02", "2021-03-02", "abc", "1", "3.0","2023-01-02"],
["1", "2.0", "3.0", "2021-03-02", "2021-03-02", "abc", "1", "3.0","2023-01-02"],
["1", "2.0", "3.0", "2021-03-02", "2021-03-02", "abc", "1", "3.0","2023-01-02"],
["2", "2", "2.0", "2021-02-03", "2021-01-02 00:01:02","efg", "", "", ""]],
Lines = [[list_to_binary(I) || I <- Row] || Row <- LineS],
Res = guess_data_types(true, Lines),
?assertEqual({[{integer, 1,0},
{number, 3,0},
{float, 3,0},
{date, 10,0},
{datetime,19,0},
{string, 3,0},
{integer, 1,1},
{float, 3,1},
{date, 10,1}],
[
[<<"a">>,<<"b">>,<<"c">>,<<"d">>,<<"e">>,<<"f">>,<<"g">>,<<"h">>,<<"i">>],
[1,1.0,1.0,{2021,01,02},{2021,01,02}, <<"abc">>, 1, 1.0, {2021,01,02}],
[1,2.0,3.0,{2021,03,02},{2021,03,02}, <<"abc">>, 1, 3.0, {2023,01,02}],
[1,2.0,3.0,{2021,03,02},{2021,03,02}, <<"abc">>, 1, 3.0, {2023,01,02}],
[1,2.0,3.0,{2021,03,02},{2021,03,02}, <<"abc">>, 1, 3.0, {2023,01,02}],
[2,2, 2.0,{2021,02,03},{{2021,01,02},{0,1,2}}, <<"efg">>, null, null, null]
]},
Res),
Res2 = guess_data_types(true, Lines, 10.0, 5),
?assertEqual({[{integer, 1,0},
{number, 3,0},
{float, 3,0},
{date, 10,0},
{datetime,19,0},
{string, 3,0},
{string, 1,1},
{string, 3,1},
{string, 10,1}],
[
[<<"a">>,<<"b">>,<<"c">>,<<"d">>,<<"e">>,<<"f">>,<<"g">>,<<"h">>,<<"i">>],
[1,1.0,1.0,{2021,01,02},{2021,01,02}, <<"abc">>, 1, 1.0, {2021,01,02}],
[1,2.0,3.0,{2021,03,02},{2021,03,02}, <<"abc">>, 1, 3.0, {2023,01,02}],
[1,2.0,3.0,{2021,03,02},{2021,03,02}, <<"abc">>, 1, 3.0, {2023,01,02}],
[1,2.0,3.0,{2021,03,02},{2021,03,02}, <<"abc">>, 1, 3.0, {2023,01,02}],
[2,2, 2.0,{2021,02,03},{{2021,01,02},{0,1,2}}, <<"efg">>, null, null, null]
]},
Res2).
-endif.
