Packages
ch
0.2.10
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/row_binary.ex
defmodule Ch.RowBinary do
@moduledoc "Helpers for working with ClickHouse [`RowBinary`](https://clickhouse.com/docs/en/sql-reference/formats#rowbinary) format."
# @compile {:bin_opt_info, true}
@dialyzer :no_improper_lists
import Bitwise
@epoch_date ~D[1970-01-01]
@epoch_naive_datetime NaiveDateTime.new!(@epoch_date, ~T[00:00:00])
@epoch_utc_datetime DateTime.new!(@epoch_date, ~T[00:00:00])
@doc false
def encode_names_and_types(names, types) do
[encode(:varint, length(names)), encode_many(names, :string), encode_types(types)]
end
defp encode_types([type | types]) do
encoded =
case type do
_ when is_binary(type) -> type
_ -> Ch.Types.encode(type)
end
[encode(:string, encoded) | encode_types(types)]
end
defp encode_types([] = done), do: done
@doc """
Encodes a single row to [`RowBinary`](https://clickhouse.com/docs/en/sql-reference/formats#rowbinary) as iodata.
Examples:
iex> encode_row([], [])
[]
iex> encode_row([1], ["UInt8"])
[1]
iex> encode_row([3, "hello"], ["UInt8", "String"])
[3, [5 | "hello"]]
"""
def encode_row(row, types) do
_encode_row(row, encoding_types(types))
end
defp _encode_row([el | els], [type | types]), do: [encode(type, el) | _encode_row(els, types)]
defp _encode_row([] = done, []), do: done
@doc """
Encodes multiple rows to [`RowBinary`](https://clickhouse.com/docs/en/sql-reference/formats#rowbinary) as iodata.
Examples:
iex> encode_rows([], [])
[]
iex> encode_rows([[1]], ["UInt8"])
[1]
iex> encode_rows([[3, "hello"], [4, "hi"]], ["UInt8", "String"])
[3, [5 | "hello"], 4, [2 | "hi"]]
"""
def encode_rows(rows, types) do
_encode_rows(rows, encoding_types(types))
end
@doc false
def _encode_rows([row | rows], types), do: _encode_rows(row, types, rows, types)
def _encode_rows([] = done, _types), do: done
defp _encode_rows([el | els], [t | ts], rows, types) do
[encode(t, el) | _encode_rows(els, ts, rows, types)]
end
defp _encode_rows([], [], rows, types), do: _encode_rows(rows, types)
@doc false
def encoding_types([type | types]) do
[encoding_type(type) | encoding_types(types)]
end
def encoding_types([] = done), do: done
defp encoding_type(type) when is_binary(type) do
encoding_type(Ch.Types.decode(type))
end
defp encoding_type(t)
when t in [
:string,
:binary,
:boolean,
:uuid,
:date,
:datetime,
:date32,
:ipv4,
:ipv6,
:point,
:nothing
],
do: t
defp encoding_type({:datetime = d, "UTC"}), do: d
defp encoding_type({:datetime, tz}) do
raise ArgumentError, "can't encode DateTime with non-UTC timezone: #{inspect(tz)}"
end
defp encoding_type({:fixed_string, _len} = t), do: t
for size <- [8, 16, 32, 64, 128, 256] do
defp encoding_type(unquote(:"u#{size}") = u), do: u
defp encoding_type(unquote(:"i#{size}") = i), do: i
end
for size <- [32, 64] do
defp encoding_type(unquote(:"f#{size}") = f), do: f
end
defp encoding_type({:array = a, t}), do: {a, encoding_type(t)}
defp encoding_type({:tuple = t, ts}) do
{t, Enum.map(ts, &encoding_type/1)}
end
defp encoding_type({:map = m, kt, vt}) do
{m, encoding_type(kt), encoding_type(vt)}
end
defp encoding_type({:nullable = n, t}), do: {n, encoding_type(t)}
defp encoding_type({:low_cardinality, t}), do: encoding_type(t)
defp encoding_type({:decimal, p, s}) do
case decimal_size(p) do
32 -> {:decimal32, s}
64 -> {:decimal64, s}
128 -> {:decimal128, s}
256 -> {:decimal256, s}
end
end
defp encoding_type({d, _scale} = t)
when d in [:decimal32, :decimal64, :decimal128, :decimal256],
do: t
defp encoding_type({:datetime64 = t, p}), do: {t, time_unit(p)}
defp encoding_type({:datetime64 = t, p, "UTC"}), do: {t, time_unit(p)}
defp encoding_type({:datetime64, _, tz}) do
raise ArgumentError, "can't encode DateTime64 with non-UTC timezone: #{inspect(tz)}"
end
defp encoding_type({e, mappings}) when e in [:enum8, :enum16] do
{e, Map.new(mappings)}
end
defp encoding_type({:simple_aggregate_function, _f, t}), do: encoding_type(t)
defp encoding_type(:ring), do: {:array, :point}
defp encoding_type(:polygon), do: {:array, {:array, :point}}
defp encoding_type(:multipolygon), do: {:array, {:array, {:array, :point}}}
defp encoding_type(type) do
raise ArgumentError, "unsupported type for encoding: #{inspect(type)}"
end
@doc false
def encode(type, value)
def encode(:varint, i) when is_integer(i) and i < 128, do: i
def encode(:varint, i) when is_integer(i), do: encode_varint_cont(i)
def encode(type, str) when type in [:string, :binary] do
case str do
_ when is_binary(str) -> [encode(:varint, byte_size(str)) | str]
_ when is_list(str) -> [encode(:varint, IO.iodata_length(str)) | str]
nil -> 0
end
end
def encode({:fixed_string, size}, str) when byte_size(str) == size do
str
end
def encode({:fixed_string, size}, str) when byte_size(str) < size do
to_pad = size - byte_size(str)
[str | <<0::size(to_pad * 8)>>]
end
def encode({:fixed_string, size}, nil), do: <<0::size(size * 8)>>
# UInt8 — [0 : 255]
def encode(:u8, u) when is_integer(u) and u >= 0 and u <= 255, do: u
def encode(:u8, nil), do: 0
def encode(:u8, term) do
raise ArgumentError, "invalid UInt8: #{inspect(term)}"
end
# Int8 — [-128 : 127]
def encode(:i8, i) when is_integer(i) and i >= 0 and i <= 127, do: i
def encode(:i8, i) when is_integer(i) and i < 0 and i >= -128, do: <<i::signed>>
def encode(:i8, nil), do: 0
def encode(:i8, term) do
raise ArgumentError, "invalid Int8: #{inspect(term)}"
end
for size <- [16, 32, 64, 128, 256] do
def encode(unquote(:"u#{size}"), u) when is_integer(u) do
<<u::unquote(size)-little>>
end
def encode(unquote(:"i#{size}"), i) when is_integer(i) do
<<i::unquote(size)-little-signed>>
end
def encode(unquote(:"u#{size}"), nil), do: <<0::unquote(size)>>
def encode(unquote(:"i#{size}"), nil), do: <<0::unquote(size)>>
end
for size <- [32, 64] do
type = :"f#{size}"
def encode(unquote(type), f) when is_number(f) do
<<f::unquote(size)-little-signed-float>>
end
def encode(unquote(type), nil), do: <<0::unquote(size)>>
end
def encode({:decimal, precision, scale}, decimal) do
type =
case decimal_size(precision) do
32 -> :decimal32
64 -> :decimal64
128 -> :decimal128
256 -> :decimal256
end
encode({type, scale}, decimal)
end
for size <- [32, 64, 128, 256] do
type = :"decimal#{size}"
def encode({unquote(type), scale} = t, %Decimal{sign: sign, coef: coef, exp: exp} = d) do
cond do
scale == -exp ->
i = sign * coef
<<i::unquote(size)-little>>
exp >= 0 ->
i = sign * coef * round(:math.pow(10, exp + scale))
<<i::unquote(size)-little>>
true ->
encode(t, Decimal.round(d, scale))
end
end
def encode({unquote(type), _scale}, nil), do: <<0::unquote(size)>>
end
def encode(:boolean, true), do: 1
def encode(:boolean, false), do: 0
def encode(:boolean, nil), do: 0
def encode({:array, type}, [_ | _] = l) do
[encode(:varint, length(l)) | encode_many(l, type)]
end
def encode({:array, _type}, []), do: 0
def encode({:array, _type}, nil), do: 0
def encode({:map, k, v}, [_ | _] = m) do
[encode(:varint, length(m)) | encode_many_kv(m, k, v)]
end
def encode({:map, _k, _v} = t, m) when is_map(m), do: encode(t, Map.to_list(m))
def encode({:map, _k, _v}, []), do: 0
def encode({:map, _k, _v}, nil), do: 0
def encode({:tuple, _types} = t, v) when is_tuple(v) do
encode(t, Tuple.to_list(v))
end
def encode({:tuple, types}, values) when is_list(types) and is_list(values) do
encode_row(values, types)
end
def encode({:tuple, types}, nil) when is_list(types) do
Enum.map(types, fn type -> encode(type, nil) end)
end
def encode(:datetime, %NaiveDateTime{} = datetime) do
<<NaiveDateTime.diff(datetime, @epoch_naive_datetime)::32-little>>
end
def encode(:datetime, %DateTime{time_zone: "Etc/UTC"} = datetime) do
<<DateTime.to_unix(datetime, :second)::32-little>>
end
def encode(:datetime, %DateTime{} = datetime) do
raise ArgumentError, "non-UTC timezones are not supported for encoding: #{datetime}"
end
def encode(:datetime, nil), do: <<0::32>>
def encode({:datetime64, time_unit}, %NaiveDateTime{} = datetime) do
<<NaiveDateTime.diff(datetime, @epoch_naive_datetime, time_unit)::64-little-signed>>
end
def encode({:datetime64, time_unit}, %DateTime{time_zone: "Etc/UTC"} = datetime) do
<<DateTime.diff(datetime, @epoch_utc_datetime, time_unit)::64-little-signed>>
end
def encode({:datetime64, _precision}, %DateTime{} = datetime) do
raise ArgumentError, "non-UTC timezones are not supported for encoding: #{datetime}"
end
def encode({:datetime64, _precision}, nil), do: <<0::64>>
def encode(:date, %Date{} = date) do
<<Date.diff(date, @epoch_date)::16-little>>
end
def encode(:date, nil), do: <<0::16>>
def encode(:date32, %Date{} = date) do
<<Date.diff(date, @epoch_date)::32-little-signed>>
end
def encode(:date32, nil), do: <<0::32>>
def encode(:uuid, <<u1::64, u2::64>>), do: <<u1::64-little, u2::64-little>>
def encode(
:uuid,
<<a1, a2, a3, a4, a5, a6, a7, a8, ?-, b1, b2, b3, b4, ?-, c1, c2, c3, c4, ?-, d1, d2, d3,
d4, ?-, e1, e2, e3, e4, e5, e6, e7, e8, e9, e10, e11, e12>>
) do
raw =
<<d(a1)::4, d(a2)::4, d(a3)::4, d(a4)::4, d(a5)::4, d(a6)::4, d(a7)::4, d(a8)::4, d(b1)::4,
d(b2)::4, d(b3)::4, d(b4)::4, d(c1)::4, d(c2)::4, d(c3)::4, d(c4)::4, d(d1)::4, d(d2)::4,
d(d3)::4, d(d4)::4, d(e1)::4, d(e2)::4, d(e3)::4, d(e4)::4, d(e5)::4, d(e6)::4, d(e7)::4,
d(e8)::4, d(e9)::4, d(e10)::4, d(e11)::4, d(e12)::4>>
encode(:uuid, raw)
end
def encode(:uuid, nil), do: <<0::128>>
def encode(:ipv4, {a, b, c, d}), do: [d, c, b, a]
def encode(:ipv4, nil), do: <<0::32>>
def encode(:ipv6, {b1, b2, b3, b4, b5, b6, b7, b8}) do
<<b1::16, b2::16, b3::16, b4::16, b5::16, b6::16, b7::16, b8::16>>
end
def encode(:ipv6, <<_::128>> = encoded), do: encoded
def encode(:ipv6, nil), do: <<0::128>>
def encode(:point, {x, y}), do: [encode(:f64, x) | encode(:f64, y)]
def encode(:point, nil), do: <<0::128>>
def encode(:ring, points), do: encode({:array, :point}, points)
def encode(:polygon, rings), do: encode({:array, :ring}, rings)
def encode(:multipolygon, polygons), do: encode({:array, :polygon}, polygons)
# TODO enum8 and enum16 nil
for size <- [8, 16] do
enum_t = :"enum#{size}"
int_t = :"i#{size}"
def encode({unquote(enum_t), mapping}, e) do
i =
case e do
_ when is_integer(e) ->
e
_ when is_binary(e) ->
case Map.fetch(mapping, e) do
{:ok, res} ->
res
:error ->
raise ArgumentError,
"enum value #{inspect(e)} not found in mapping: #{inspect(mapping)}"
end
end
encode(unquote(int_t), i)
end
end
def encode({:nullable, _type}, nil), do: 1
def encode({:nullable, type}, value) do
case encode(type, value) do
e when is_list(e) or is_binary(e) -> [0 | e]
e -> [0, e]
end
end
defp encode_varint_cont(i) when i < 128, do: <<i>>
defp encode_varint_cont(i) do
[(i &&& 0b0111_1111) ||| 0b1000_0000 | encode_varint_cont(i >>> 7)]
end
defp encode_many([el | rest], type), do: [encode(type, el) | encode_many(rest, type)]
defp encode_many([] = done, _type), do: done
defp encode_many_kv([{key, value} | rest], key_type, value_type) do
[
encode(key_type, key),
encode(value_type, value)
| encode_many_kv(rest, key_type, value_type)
]
end
defp encode_many_kv([] = done, _key_type, _value_type), do: done
@compile {:inline, d: 1}
defp d(?0), do: 0
defp d(?1), do: 1
defp d(?2), do: 2
defp d(?3), do: 3
defp d(?4), do: 4
defp d(?5), do: 5
defp d(?6), do: 6
defp d(?7), do: 7
defp d(?8), do: 8
defp d(?9), do: 9
defp d(?A), do: 10
defp d(?B), do: 11
defp d(?C), do: 12
defp d(?D), do: 13
defp d(?E), do: 14
defp d(?F), do: 15
defp d(?a), do: 10
defp d(?b), do: 11
defp d(?c), do: 12
defp d(?d), do: 13
defp d(?e), do: 14
defp d(?f), do: 15
@doc """
Decodes [`RowBinaryWithNamesAndTypes`](https://clickhouse.com/docs/en/sql-reference/formats#rowbinarywithnamesandtypes) into rows.
Example:
iex> decode_rows(<<1, 3, "1+1"::bytes, 5, "UInt8"::bytes, 2>>)
[[2]]
"""
def decode_rows(row_binary_with_names_and_types)
def decode_rows(<<cols, rest::bytes>>), do: skip_names(rest, cols, cols)
def decode_rows(<<>>), do: []
@doc """
Decodes [`RowBinary`](https://clickhouse.com/docs/en/sql-reference/formats#rowbinary) into rows.
Example:
iex> decode_rows(<<1>>, ["UInt8"])
[[1]]
"""
def decode_rows(row_binary, types)
def decode_rows(<<>>, _types), do: []
def decode_rows(<<data::bytes>>, types) do
types = decoding_types(types)
decode_rows(types, data, [], [], types)
end
@doc false
def decoding_types([type | types]) do
[decoding_type(type) | types]
end
def decoding_types([] = done), do: done
defp decoding_type(t) when is_binary(t) do
decoding_type(Ch.Types.decode(t))
end
defp decoding_type(t)
when t in [
:string,
:binary,
:boolean,
:uuid,
:date,
:date32,
:ipv4,
:ipv6,
:point,
:nothing
],
do: t
defp decoding_type({:datetime, _tz} = t), do: t
defp decoding_type({:fixed_string, _len} = t), do: t
for size <- [8, 16, 32, 64, 128, 256] do
defp decoding_type(unquote(:"u#{size}") = u), do: u
defp decoding_type(unquote(:"i#{size}") = i), do: i
end
for size <- [32, 64] do
defp decoding_type(unquote(:"f#{size}") = f), do: f
end
defp decoding_type(:datetime = t), do: {t, _tz = nil}
defp decoding_type({:array = a, t}), do: {a, decoding_type(t)}
defp decoding_type({:tuple = t, ts}) do
{t, Enum.map(ts, &decoding_type/1)}
end
defp decoding_type({:map = m, kt, vt}) do
{m, decoding_type(kt), decoding_type(vt)}
end
defp decoding_type({:nullable = n, t}), do: {n, decoding_type(t)}
defp decoding_type({:low_cardinality, t}), do: decoding_type(t)
defp decoding_type({:decimal = t, p, s}), do: {t, decimal_size(p), s}
defp decoding_type({:decimal32, s}), do: {:decimal, 32, s}
defp decoding_type({:decimal64, s}), do: {:decimal, 64, s}
defp decoding_type({:decimal128, s}), do: {:decimal, 128, s}
defp decoding_type({:decimal256, s}), do: {:decimal, 256, s}
defp decoding_type({:datetime64 = t, p}), do: {t, time_unit(p), _tz = nil}
defp decoding_type({:datetime64 = t, p, tz}), do: {t, time_unit(p), tz}
defp decoding_type({e, mappings}) when e in [:enum8, :enum16] do
{e, Map.new(mappings, fn {k, v} -> {v, k} end)}
end
defp decoding_type({:simple_aggregate_function, _f, t}), do: decoding_type(t)
defp decoding_type(:ring), do: {:array, :point}
defp decoding_type(:polygon), do: {:array, {:array, :point}}
defp decoding_type(:multipolygon), do: {:array, {:array, {:array, :point}}}
defp decoding_type(type) do
raise ArgumentError, "unsupported type for decoding: #{inspect(type)}"
end
defp skip_names(<<rest::bytes>>, 0, count), do: decode_types(rest, count, _acc = [])
varints = [
{_pattern = quote(do: <<0::1, v1::7>>), _value = quote(do: v1)},
{quote(do: <<1::1, v1::7, 0::1, v2::7>>), quote(do: (v2 <<< 7) + v1)},
{quote(do: <<1::1, v1::7, 1::1, v2::7, 0::1, v3::7>>),
quote(do: (v3 <<< 14) + (v2 <<< 7) + v1)},
{quote(do: <<1::1, v1::7, 1::1, v2::7, 1::1, v3::7, 0::1, v4::7>>),
quote(do: (v4 <<< 21) + (v3 <<< 14) + (v2 <<< 7) + v1)},
{quote(do: <<1::1, v1::7, 1::1, v2::7, 1::1, v3::7, 1::1, v4::7, 0::1, v5::7>>),
quote(do: (v5 <<< 28) + (v4 <<< 21) + (v3 <<< 14) + (v2 <<< 7) + v1)},
{quote(do: <<1::1, v1::7, 1::1, v2::7, 1::1, v3::7, 1::1, v4::7, 1::1, v5::7, 0::1, v6::7>>),
quote(do: (v6 <<< 35) + (v5 <<< 28) + (v4 <<< 21) + (v3 <<< 14) + (v2 <<< 7) + v1)},
{quote do
<<1::1, v1::7, 1::1, v2::7, 1::1, v3::7, 1::1, v4::7, 1::1, v5::7, 1::1, v6::7, 0::1,
v7::7>>
end,
quote do
(v7 <<< 42) + (v6 <<< 35) + (v5 <<< 28) + (v4 <<< 21) + (v3 <<< 14) + (v2 <<< 7) + v1
end},
{quote do
<<1::1, v1::7, 1::1, v2::7, 1::1, v3::7, 1::1, v4::7, 1::1, v5::7, 1::1, v6::7, 1::1,
v7::7, 0::1, v8::7>>
end,
quote do
(v8 <<< 49) + (v7 <<< 42) + (v6 <<< 35) + (v5 <<< 28) + (v4 <<< 21) + (v3 <<< 14) +
(v2 <<< 7) + v1
end}
]
for {pattern, value} <- varints do
defp skip_names(<<unquote(pattern), _::size(unquote(value))-bytes, rest::bytes>>, left, count) do
skip_names(rest, left - 1, count)
end
end
defp decode_types(<<>>, 0, _types), do: []
defp decode_types(<<rest::bytes>>, 0, types) do
types = types |> decode_types() |> :lists.reverse()
decode_rows(types, rest, _row = [], _rows = [], types)
end
defp decode_types(<<size, type::size(size)-bytes, rest::bytes>>, count, acc) do
decode_types(rest, count - 1, [type | acc])
end
@doc false
def decode_types([type | types]) do
[decoding_type(Ch.Types.decode(type)) | decode_types(types)]
end
def decode_types([] = done), do: done
@compile inline: [decode_string_decode_rows: 5]
for {pattern, size} <- varints do
defp decode_string_decode_rows(
<<unquote(pattern), s::size(unquote(size))-bytes, bin::bytes>>,
types_rest,
row,
rows,
types
) do
decode_rows(types_rest, bin, [to_utf8(s) | row], rows, types)
end
end
@doc false
def to_utf8(str) do
utf8 = to_utf8(str, 0, 0, str, [])
IO.iodata_to_binary(utf8)
end
@dialyzer {:no_improper_lists, to_utf8: 5, to_utf8_escape: 5}
defp to_utf8(<<valid::utf8, rest::bytes>>, from, len, original, acc) do
to_utf8(rest, from, len + utf8_size(valid), original, acc)
end
defp to_utf8(<<_invalid, rest::bytes>>, from, len, original, acc) do
acc = [acc | binary_part(original, from, len)]
to_utf8_escape(rest, from + len, 1, original, acc)
end
defp to_utf8(<<>>, from, len, original, acc) do
[acc | binary_part(original, from, len)]
end
defp to_utf8_escape(<<valid::utf8, rest::bytes>>, from, len, original, acc) do
acc = [acc | "�"]
to_utf8(rest, from + len, utf8_size(valid), original, acc)
end
defp to_utf8_escape(<<_invalid, rest::bytes>>, from, len, original, acc) do
to_utf8_escape(rest, from, len + 1, original, acc)
end
defp to_utf8_escape(<<>>, _from, _len, _original, acc) do
[acc | "�"]
end
# UTF-8 encodes code points in one to four bytes
@compile inline: [utf8_size: 1]
defp utf8_size(codepoint) when codepoint <= 0x7F, do: 1
defp utf8_size(codepoint) when codepoint <= 0x7FF, do: 2
defp utf8_size(codepoint) when codepoint <= 0xFFFF, do: 3
defp utf8_size(codepoint) when codepoint <= 0x10FFFF, do: 4
@compile inline: [decode_binary_decode_rows: 5]
for {pattern, size} <- varints do
defp decode_binary_decode_rows(
<<unquote(pattern), s::size(unquote(size))-bytes, bin::bytes>>,
types_rest,
row,
rows,
types
) do
decode_rows(types_rest, bin, [s | row], rows, types)
end
end
@compile inline: [decode_array_decode_rows: 6]
defp decode_array_decode_rows(<<0, bin::bytes>>, _type, types_rest, row, rows, types) do
decode_rows(types_rest, bin, [[] | row], rows, types)
end
for {pattern, size} <- varints do
defp decode_array_decode_rows(
<<unquote(pattern), bin::bytes>>,
type,
types_rest,
row,
rows,
types
) do
array_types = List.duplicate(type, unquote(size))
types_rest = array_types ++ [{:array_over, row} | types_rest]
decode_rows(types_rest, bin, [], rows, types)
end
end
@compile inline: [decode_map_decode_rows: 7]
defp decode_map_decode_rows(
<<0, bin::bytes>>,
_key_type,
_value_type,
types_rest,
row,
rows,
types
) do
decode_rows(types_rest, bin, [%{} | row], rows, types)
end
for {pattern, size} <- varints do
defp decode_map_decode_rows(
<<unquote(pattern), bin::bytes>>,
key_type,
value_type,
types_rest,
row,
rows,
types
) do
types_rest =
map_types(unquote(size), key_type, value_type) ++ [{:map_over, row} | types_rest]
decode_rows(types_rest, bin, [], rows, types)
end
end
defp map_types(count, key_type, value_type) when count > 0 do
[key_type, value_type | map_types(count - 1, key_type, value_type)]
end
defp map_types(0, _key_type, _value_types), do: []
defp decode_rows([type | types_rest], <<bin::bytes>>, row, rows, types) do
case type do
:u8 ->
<<u, bin::bytes>> = bin
decode_rows(types_rest, bin, [u | row], rows, types)
:u16 ->
<<u::16-little, bin::bytes>> = bin
decode_rows(types_rest, bin, [u | row], rows, types)
:u32 ->
<<u::32-little, bin::bytes>> = bin
decode_rows(types_rest, bin, [u | row], rows, types)
:u64 ->
<<u::64-little, bin::bytes>> = bin
decode_rows(types_rest, bin, [u | row], rows, types)
:u128 ->
<<u::128-little, bin::bytes>> = bin
decode_rows(types_rest, bin, [u | row], rows, types)
:u256 ->
<<u::256-little, bin::bytes>> = bin
decode_rows(types_rest, bin, [u | row], rows, types)
:i8 ->
<<i::signed, bin::bytes>> = bin
decode_rows(types_rest, bin, [i | row], rows, types)
:i16 ->
<<i::16-little-signed, bin::bytes>> = bin
decode_rows(types_rest, bin, [i | row], rows, types)
:i32 ->
<<i::32-little-signed, bin::bytes>> = bin
decode_rows(types_rest, bin, [i | row], rows, types)
:i64 ->
<<i::64-little-signed, bin::bytes>> = bin
decode_rows(types_rest, bin, [i | row], rows, types)
:i128 ->
<<i::128-little-signed, bin::bytes>> = bin
decode_rows(types_rest, bin, [i | row], rows, types)
:i256 ->
<<i::256-little-signed, bin::bytes>> = bin
decode_rows(types_rest, bin, [i | row], rows, types)
:f32 ->
case bin do
<<f::32-little-float, bin::bytes>> ->
decode_rows(types_rest, bin, [f | row], rows, types)
<<_nan_or_inf::32, bin::bytes>> ->
decode_rows(types_rest, bin, [nil | row], rows, types)
end
:f64 ->
case bin do
<<f::64-little-float, bin::bytes>> ->
decode_rows(types_rest, bin, [f | row], rows, types)
<<_nan_or_inf::64, bin::bytes>> ->
decode_rows(types_rest, bin, [nil | row], rows, types)
end
:string ->
decode_string_decode_rows(bin, types_rest, row, rows, types)
:binary ->
decode_binary_decode_rows(bin, types_rest, row, rows, types)
# TODO utf8?
{:fixed_string, size} ->
<<s::size(size)-bytes, bin::bytes>> = bin
decode_rows(types_rest, bin, [s | row], rows, types)
:boolean ->
case bin do
<<0, bin::bytes>> -> decode_rows(types_rest, bin, [false | row], rows, types)
<<1, bin::bytes>> -> decode_rows(types_rest, bin, [true | row], rows, types)
end
:uuid ->
<<u1::64-little, u2::64-little, bin::bytes>> = bin
uuid = <<u1::64, u2::64>>
decode_rows(types_rest, bin, [uuid | row], rows, types)
:date ->
<<d::16-little, bin::bytes>> = bin
decode_rows(types_rest, bin, [Date.add(@epoch_date, d) | row], rows, types)
:date32 ->
<<d::32-little-signed, bin::bytes>> = bin
decode_rows(types_rest, bin, [Date.add(@epoch_date, d) | row], rows, types)
{:datetime, timezone} ->
<<s::32-little, bin::bytes>> = bin
dt =
case timezone do
nil -> NaiveDateTime.add(@epoch_naive_datetime, s)
"UTC" -> DateTime.from_unix!(s)
_ -> s |> DateTime.from_unix!() |> DateTime.shift_zone!(timezone)
end
decode_rows(types_rest, bin, [dt | row], rows, types)
{:decimal, size, scale} ->
<<val::size(size)-little-signed, bin::bytes>> = bin
sign = if val < 0, do: -1, else: 1
d = Decimal.new(sign, abs(val), -scale)
decode_rows(types_rest, bin, [d | row], rows, types)
{:nullable, type} ->
case bin do
<<1, bin::bytes>> -> decode_rows(types_rest, bin, [nil | row], rows, types)
<<0, bin::bytes>> -> decode_rows([type | types_rest], bin, row, rows, types)
end
{:array, type} ->
decode_array_decode_rows(bin, type, types_rest, row, rows, types)
{:array_over, original_row} ->
decode_rows(types_rest, bin, [:lists.reverse(row) | original_row], rows, types)
{:map, key_type, value_type} ->
decode_map_decode_rows(bin, key_type, value_type, types_rest, row, rows, types)
{:map_over, original_row} ->
map = row |> Enum.chunk_every(2) |> Enum.map(fn [v, k] -> {k, v} end) |> Map.new()
decode_rows(types_rest, bin, [map | original_row], rows, types)
{:tuple, tuple_types} ->
decode_rows(tuple_types ++ [{:tuple_over, row} | types_rest], bin, [], rows, types)
{:tuple_over, original_row} ->
tuple = row |> :lists.reverse() |> List.to_tuple()
decode_rows(types_rest, bin, [tuple | original_row], rows, types)
{:datetime64, time_unit, timezone} ->
<<s::64-little-signed, bin::bytes>> = bin
dt =
case timezone do
nil ->
NaiveDateTime.add(@epoch_naive_datetime, s, time_unit)
"UTC" ->
DateTime.from_unix!(s, time_unit)
_ ->
s
|> DateTime.from_unix!(time_unit)
|> DateTime.shift_zone!(timezone)
end
decode_rows(types_rest, bin, [dt | row], rows, types)
{:enum8, mapping} ->
<<v::signed, bin::bytes>> = bin
decode_rows(types_rest, bin, [Map.fetch!(mapping, v) | row], rows, types)
{:enum16, mapping} ->
<<v::16-little-signed, bin::bytes>> = bin
decode_rows(types_rest, bin, [Map.fetch!(mapping, v) | row], rows, types)
:ipv4 ->
<<b4, b3, b2, b1, bin::bytes>> = bin
decode_rows(types_rest, bin, [{b1, b2, b3, b4} | row], rows, types)
:ipv6 ->
<<b1::16, b2::16, b3::16, b4::16, b5::16, b6::16, b7::16, b8::16, bin::bytes>> = bin
decode_rows(types_rest, bin, [{b1, b2, b3, b4, b5, b6, b7, b8} | row], rows, types)
:point ->
<<x::64-little-float, y::64-little-float, bin::bytes>> = bin
decode_rows(types_rest, bin, [{x, y} | row], rows, types)
end
end
defp decode_rows([], <<>>, row, rows, _types) do
:lists.reverse([:lists.reverse(row) | rows])
end
defp decode_rows([], <<bin::bytes>>, row, rows, types) do
row = :lists.reverse(row)
decode_rows(types, bin, [], [row | rows], types)
end
@compile inline: [decimal_size: 1]
# https://clickhouse.com/docs/en/sql-reference/data-types/decimal/
defp decimal_size(precision) when is_integer(precision) do
cond do
precision >= 39 -> 256
precision >= 19 -> 128
precision >= 10 -> 64
true -> 32
end
end
@compile inline: [time_unit: 1]
for precision <- 0..9 do
time_unit = round(:math.pow(10, precision))
defp time_unit(unquote(precision)), do: unquote(time_unit)
end
end