Packages
ch
0.8.0
0.8.3
0.8.2
0.8.1
0.8.0
0.7.3
0.7.2
0.7.1
0.7.0
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.9
0.5.8
0.5.7
0.5.6
0.5.5
0.5.4
retired
0.5.3
retired
0.5.2
retired
0.5.1
retired
0.5.0
retired
0.4.3
0.4.2
0.4.1
0.4.0
0.3.6
0.3.5
0.3.4
0.3.3
0.3.2
0.3.1
0.3.0
0.2.10
0.2.9
0.2.8
0.2.7
0.2.6
0.2.5
0.2.5-rc.0
0.2.4
0.2.3
0.2.2
0.2.2-rc.3
0.2.2-rc.2
0.2.2-rc.1
0.2.2-rc.0
0.2.1
0.2.0
0.1.15
0.1.14
0.1.13
0.1.12
0.1.11
0.1.10
0.1.9
0.1.8
0.1.7
retired
0.1.6
retired
0.1.5
retired
0.1.4
retired
0.1.3
retired
0.1.2
retired
0.1.1
0.1.0
HTTP ClickHouse driver for Elixir
Current section
Files
Jump to
Current section
Files
lib/ch/query.ex
defmodule Ch.Query do
@moduledoc "Query struct wrapping the SQL statement."
defstruct [:statement, :command, :encode, :decode, :multipart]
@typedoc """
The Query struct.
## Fields
* `:statement` - The SQL statement to be executed (as `t:iodata/0`).
* `:command` - The detected or enforced SQL command type (e.g., `:select`, `:insert`).
* `:encode` - Whether to encode parameters (defaults to `true`).
* `:decode` - Whether to decode the response (defaults to `true`).
* `:multipart` - Whether to use `multipart/form-data` for the request (defaults to `false`).
"""
@type t :: %__MODULE__{
statement: iodata,
command: command,
encode: boolean,
decode: boolean,
multipart: boolean
}
@doc false
@spec build(iodata, [Ch.query_option()]) :: t
def build(statement, opts \\ []) do
command = Keyword.get(opts, :command) || extract_command(statement)
encode = Keyword.get(opts, :encode, true)
decode = Keyword.get(opts, :decode, true)
multipart = Keyword.get(opts, :multipart, false)
%__MODULE__{
statement: statement,
command: command,
encode: encode,
decode: decode,
multipart: multipart
}
end
statements = [
{"SELECT", :select},
{"INSERT", :insert},
{"CREATE", :create},
{"ALTER", :alter},
{"DELETE", :delete},
{"SYSTEM", :system},
{"SHOW", :show},
# as of ClickHouse 24.11, WITH is only allowed in SELECT
# https://clickhouse.com/docs/en/sql-reference/statements/select/with/
{"WITH", :select},
{"GRANT", :grant},
{"EXPLAIN", :explain},
{"REVOKE", :revoke},
{"UPDATE", :update},
{"ATTACH", :attach},
{"CHECK", :check},
{"DESCRIBE", :describe},
{"DETACH", :detach},
{"DROP", :drop},
{"EXISTS", :exists},
{"KILL", :kill},
{"OPTIMIZE", :optimize},
{"RENAME", :rename},
{"EXCHANGE", :exchange},
{"SET", :set},
{"TRUNCATE", :truncate},
{"USE", :use},
{"WATCH", :watch},
{"MOVE", :move},
{"UNDROP", :undrop}
]
command_union =
statements
|> Enum.map(fn {_, command} -> command end)
|> Enum.reduce(&{:|, [], [&1, &2]})
@typedoc """
Atom representing the type of SQL command.
Derived automatically from the start of the SQL statement (e.g., `"SELECT ..."` -> `:select`),
or provided explicitly via options.
"""
@type command :: unquote(command_union)
defp extract_command(statement)
for {statement, command} <- statements do
defp extract_command(unquote(statement) <> _), do: unquote(command)
defp extract_command(unquote(String.downcase(statement)) <> _), do: unquote(command)
end
defp extract_command(<<whitespace, rest::bytes>>) when whitespace in [?\s, ?\t, ?\n] do
extract_command(rest)
end
defp extract_command([first_segment | _] = statement) do
extract_command(first_segment) || extract_command(IO.iodata_to_binary(statement))
end
defp extract_command(_other), do: nil
end
defimpl DBConnection.Query, for: Ch.Query do
@dialyzer :no_improper_lists
alias Ch.{Query, Result, RowBinary}
@spec parse(Query.t(), [Ch.query_option()]) :: Query.t()
def parse(query, _opts), do: query
@spec describe(Query.t(), [Ch.query_option()]) :: Query.t()
def describe(query, _opts), do: query
# stream: insert init
@spec encode(Query.t(), {:stream, term}, [Ch.query_option()]) ::
{:stream, {[{String.t(), String.t()}], Mint.Types.headers(), iodata}}
def encode(query, {:stream, params}, opts) do
{:stream, encode(query, params, opts)}
end
# stream: insert data chunk
@spec encode(Query.t(), {:stream, Mint.Types.request_ref(), iodata | :eof}, [Ch.query_option()]) ::
{:stream, Mint.Types.request_ref(), iodata | :eof}
def encode(_query, {:stream, ref, data}, _opts) do
{:stream, ref, data}
end
@spec encode(Query.t(), params, [Ch.query_option()]) ::
{query_params, Mint.Types.headers(), body}
when params: map | [term] | [row :: [term]] | iodata | Enumerable.t(),
query_params: [{String.t(), String.t()}],
body: iodata | Enumerable.t()
def encode(%Query{command: :insert, encode: false, statement: statement}, data, opts) do
body =
case data do
_ when is_list(data) or is_binary(data) -> [statement, ?\n | data]
_ -> Stream.concat([[statement, ?\n]], data)
end
{_query_params = [], headers(opts), body}
end
def encode(%Query{command: :insert, statement: statement}, params, opts) do
cond do
names = Keyword.get(opts, :names) ->
types = Keyword.fetch!(opts, :types)
header = RowBinary.encode_names_and_types(names, types)
data = RowBinary.encode_rows(params, types)
{_query_params = [], headers(opts), [statement, ?\n, header | data]}
format_row_binary?(statement) ->
types = Keyword.fetch!(opts, :types)
data = RowBinary.encode_rows(params, types)
{_query_params = [], headers(opts), [statement, ?\n | data]}
true ->
{query_params(params), headers(opts), statement}
end
end
def encode(%Query{multipart: true, statement: statement}, params, opts) do
types = Keyword.get(opts, :types)
default_format = if types, do: "RowBinary", else: "RowBinaryWithNamesAndTypes"
format = Keyword.get(opts, :format) || default_format
boundary = "ChFormBoundary" <> Base.url_encode64(:crypto.strong_rand_bytes(24))
content_type = "multipart/form-data; boundary=\"#{boundary}\""
enc_boundary = "--#{boundary}\r\n"
multipart = multipart_params(params, enc_boundary)
multipart = add_multipart_part(multipart, "query", statement, enc_boundary)
multipart = [multipart | "--#{boundary}--\r\n"]
{_no_query_params = [],
[{"x-clickhouse-format", format}, {"content-type", content_type} | headers(opts)], multipart}
end
def encode(%Query{statement: statement}, params, opts) do
types = Keyword.get(opts, :types)
default_format = if types, do: "RowBinary", else: "RowBinaryWithNamesAndTypes"
format = Keyword.get(opts, :format) || default_format
{query_params(params), [{"x-clickhouse-format", format} | headers(opts)], statement}
end
defp multipart_params(params, boundary) when is_map(params) do
multipart_named_params(Map.to_list(params), boundary, [])
end
defp multipart_params(params, boundary) when is_list(params) do
multipart_positional_params(params, 0, boundary, [])
end
defp multipart_named_params([{name, value} | params], boundary, acc) do
acc =
add_multipart_part(
acc,
"param_" <> URI.encode_www_form(name),
encode_param(value),
boundary
)
multipart_named_params(params, boundary, acc)
end
defp multipart_named_params([], _boundary, acc), do: acc
defp multipart_positional_params([value | params], idx, boundary, acc) do
acc =
add_multipart_part(
acc,
"param_$" <> Integer.to_string(idx),
encode_param(value),
boundary
)
multipart_positional_params(params, idx + 1, boundary, acc)
end
defp multipart_positional_params([], _idx, _boundary, acc), do: acc
@compile inline: [add_multipart_part: 4]
defp add_multipart_part(multipart, name, value, boundary) do
part = [
boundary,
"content-disposition: form-data; name=\"",
name,
"\"\r\n\r\n",
value,
"\r\n"
]
case multipart do
[] -> part
_ -> [multipart | part]
end
end
defp format_row_binary?(statement) when is_binary(statement) do
statement |> String.trim_trailing() |> String.ends_with?("RowBinary")
end
defp format_row_binary?(statement) when is_list(statement) do
statement
|> IO.iodata_to_binary()
|> format_row_binary?()
end
# stream: select result
@spec decode(Query.t(), result, [Ch.query_option()]) :: result when result: Result.t()
def decode(_query, %Result{} = result, _opts), do: result
# stream: insert result
@spec decode(Query.t(), ref, [Ch.query_option()]) :: ref when ref: Mint.Types.request_ref()
def decode(_query, ref, _opts) when is_reference(ref), do: ref
@spec decode(Query.t(), [response], [Ch.query_option()]) :: Result.t()
when response: Mint.Types.status() | Mint.Types.headers() | binary
def decode(%Query{command: :insert}, responses, _opts) do
[_status, headers | _data] = responses
num_rows =
if summary = get_header(headers, "x-clickhouse-summary") do
summary = Jason.decode!(summary)
if written_rows = Map.get(summary, "written_rows") do
String.to_integer(written_rows)
end
end
%Result{num_rows: num_rows, rows: nil, command: :insert, headers: headers}
end
def decode(%Query{decode: false, command: command}, responses, _opts) when is_list(responses) do
# TODO potentially fails on x-progress-headers
[_status, headers | data] = responses
%Result{rows: data, data: data, command: command, headers: headers}
end
def decode(%Query{command: command}, responses, opts) when is_list(responses) do
# TODO potentially fails on x-progress-headers
[_status, headers | data] = responses
case get_header(headers, "x-clickhouse-format") do
"RowBinary" ->
types = Keyword.fetch!(opts, :types)
rows = data |> IO.iodata_to_binary() |> RowBinary.decode_rows(types)
%Result{num_rows: length(rows), rows: rows, command: command, headers: headers}
"RowBinaryWithNamesAndTypes" ->
[names | rows] = data |> IO.iodata_to_binary() |> RowBinary.decode_names_and_rows()
%Result{
num_rows: length(rows),
columns: names,
rows: rows,
command: command,
headers: headers
}
_other ->
%Result{rows: data, data: data, command: command, headers: headers}
end
end
defp get_header(headers, key) do
case List.keyfind(headers, key, 0) do
{_, value} -> value
nil = not_found -> not_found
end
end
defp query_params(params) when is_map(params) do
Enum.map(params, fn {k, v} -> {"param_#{k}", encode_param(v)} end)
end
defp query_params(params) when is_list(params) do
params
|> Enum.with_index()
|> Enum.map(fn {v, idx} -> {"param_$#{idx}", encode_param(v)} end)
end
defp encode_param(n) when is_integer(n), do: Integer.to_string(n)
defp encode_param(f) when is_float(f), do: Float.to_string(f)
# TODO possibly speed up
# For more info see
# https://clickhouse.com/docs/en/interfaces/http#tabs-in-url-parameters
# "escaped" format is the same as https://clickhouse.com/docs/en/interfaces/formats#tabseparated-data-formatting
defp encode_param(b) when is_binary(b) do
escape_param([{"\\", "\\\\"}, {"\t", "\\\t"}, {"\n", "\\\n"}], b)
end
defp encode_param(b) when is_boolean(b), do: Atom.to_string(b)
defp encode_param(nil), do: "\\N"
defp encode_param(%Decimal{} = d), do: Decimal.to_string(d, :normal)
defp encode_param(%Date{} = date), do: Date.to_iso8601(date)
defp encode_param(%NaiveDateTime{} = naive), do: NaiveDateTime.to_iso8601(naive)
defp encode_param(%Time{} = time), do: Time.to_iso8601(time)
defp encode_param(%DateTime{microsecond: microsecond} = dt) do
dt = DateTime.shift_zone!(dt, "Etc/UTC")
case microsecond do
{val, precision} when val > 0 and precision > 0 ->
size = Integer.pow(10, precision)
unix = DateTime.to_unix(dt, size)
seconds = div(unix, size)
fractional = rem(unix, size)
IO.iodata_to_binary([
Integer.to_string(seconds),
?.,
String.pad_leading(Integer.to_string(fractional), precision, "0")
])
_ ->
dt |> DateTime.to_unix(:second) |> Integer.to_string()
end
end
defp encode_param(tuple) when is_tuple(tuple) do
IO.iodata_to_binary([?(, encode_array_params(Tuple.to_list(tuple)), ?)])
end
defp encode_param(a) when is_list(a) do
IO.iodata_to_binary([?[, encode_array_params(a), ?]])
end
defp encode_param(m) when is_map(m) do
IO.iodata_to_binary([?{, encode_map_params(Map.to_list(m)), ?}])
end
defp encode_array_params([last]), do: encode_array_param(last)
defp encode_array_params([s | rest]) do
[encode_array_param(s), ?, | encode_array_params(rest)]
end
defp encode_array_params([] = empty), do: empty
defp encode_map_params([last]), do: encode_map_param(last)
defp encode_map_params([kv | rest]) do
[encode_map_param(kv), ?, | encode_map_params(rest)]
end
defp encode_map_params([] = empty), do: empty
defp encode_array_param(s) when is_binary(s) do
[?', escape_param([{"'", "''"}, {"\\", "\\\\"}], s), ?']
end
defp encode_array_param(nil), do: "null"
defp encode_array_param(%s{} = param) when s in [Date, NaiveDateTime] do
[?', encode_param(param), ?']
end
defp encode_array_param(v), do: encode_param(v)
defp encode_map_param({k, v}) do
[encode_array_param(k), ?:, encode_array_param(v)]
end
defp escape_param([{pattern, replacement} | escapes], param) do
param = String.replace(param, pattern, replacement)
escape_param(escapes, param)
end
defp escape_param([], param), do: param
@spec headers(Keyword.t()) :: Mint.Types.headers()
defp headers(opts), do: Keyword.get(opts, :headers, [])
end
defimpl String.Chars, for: Ch.Query do
def to_string(%{statement: statement}) do
IO.iodata_to_binary(statement)
end
end