defmodule Ch.RowBinary do @moduledoc false # @compile {:bin_opt_info, true} @dialyzer :no_improper_lists import Bitwise require Record Record.defrecord(:string, [:size]) Record.defrecord(:decimal, [:size, :scale]) Record.defrecord(:datetime64, [:unit]) @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]) def encode_row([el | els], [type | types]), do: [encode(type, el) | encode_row(els, types)] def encode_row([] = done, []), do: done 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) def encode(:varint, num) when is_integer(num) and num < 128, do: <> def encode(:varint, num) when is_integer(num) do [<<1::1, num::7>> | encode(:varint, num >>> 7)] end 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(string(size: size), str) when byte_size(str) == size do str end def encode(string(size: size), str) when byte_size(str) < size do to_pad = size - byte_size(str) [str | <<0::size(to_pad * 8)>>] end def encode(string(size: size), nil), do: <<0::size(size * 8)>> for size <- [8, 16, 32, 64, 128, 256] do def encode(unquote(:"u#{size}"), i) when is_integer(i) do <> end def encode(unquote(:"i#{size}"), i) when is_integer(i) do <> 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 def encode(unquote(:"f#{size}"), f) when is_number(f) do <> end def encode(unquote(:"f#{size}"), nil), do: <<0::unquote(size)>> end def encode(decimal(size: size, scale: scale), %Decimal{sign: sign, coef: coef, exp: exp}) when scale == -exp do i = sign * coef <> end def encode(decimal(size: size, scale: scale), %Decimal{sign: sign, coef: coef, exp: exp}) when exp >= 0 do i = sign * coef * round(:math.pow(10, exp + scale)) <> end def encode(decimal(scale: scale) = t, %Decimal{} = d) do encode(t, Decimal.round(d, scale)) end def encode(decimal(size: size), nil), do: <<0::size(size)>> 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(:datetime, %NaiveDateTime{} = datetime) do <> end def encode(:datetime, %DateTime{} = datetime) do <> end def encode(:datetime, nil), do: <<0::32>> # TODO right now the timezones are ignored during encoding # assuming the user has provided the correct one def encode({:datetime = t, _timezone}, v) do encode(t, v) end def encode(datetime64(unit: unit), %NaiveDateTime{} = datetime) do <> end def encode(datetime64(unit: unit), %DateTime{} = datetime) do <> end def encode(datetime64(), nil), do: <<0::64>> def encode(:date, %Date{} = date) do <> end def encode(:date, nil), do: <<0::16>> def encode(:date32, %Date{} = date) do <> end def encode(:date32, nil), do: <<0::32>> def encode(:uuid, <>), do: <> def encode( :uuid, <> ) do raw = <> encode(:uuid, raw) end def encode(:uuid, nil), do: <<0::128>> def encode({:nullable, _type}, nil), do: 1 def encode({:nullable, type}, value), do: [0 | encode(type, value)] defp encode_many([el | rest], type), do: [encode(type, el) | encode_many(rest, type)] defp encode_many([] = done, _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 scalar_types = [ {"String", :string}, {"UUID", :uuid}, {"UInt8", :u8}, {"UInt16", :u16}, {"UInt32", :u32}, {"UInt64", :u64}, {"UInt128", :u128}, {"UInt256", :u256}, {"Int8", :i8}, {"Int16", :i16}, {"Int32", :i32}, {"Int64", :i64}, {"Int128", :i128}, {"Int256", :i256}, {"Float32", :f32}, {"Float64", :f64}, {"Date32", :date32}, {"Bool", :boolean}, {"IPv4", :ipv4}, {"IPv6", :ipv6}, {"Nothing", :nothing} ] for {encoded, decoded} <- scalar_types do for decoded <- List.wrap(decoded) do def encode_type(unquote(decoded)), do: unquote(encoded) end end def encode_type(:binary), do: "String" def encode_type({:nullable, type}), do: ["Nullable(", encode_type(type), ?)] def encode_type({:array, type}), do: ["Array(", encode_type(type), ?)] def encode_type(:datetime), do: "DateTime" def encode_type({:datetime, timezone}) when is_binary(timezone) do ["DateTime('", timezone, "')"] end def encode_type({:datetime, nil}), do: "DateTime" def encode_type({:string, size}), do: ["FixedString(", String.Chars.Integer.to_string(size), ?)] def encode_type(:date), do: "Date" # TODO verify with custom precision Decimals for {size, precision} <- [{32, 9}, {64, 18}, {128, 38}, {256, 76}] do def encode_type(decimal(size: unquote(size), scale: scale)) do [ unquote("Decimal(#{precision}, "), String.Chars.Integer.to_string(scale), ?) ] end end # TODO datetime64, enum, etc. def decode_rows(<>), do: skip_names(rest, cols, cols) def decode_rows(<<>>), do: [] def decode_rows(<<>>, _types), do: [] def decode_rows(<>, types) do decode_rows(types, data, [], [], types) end defp skip_names(<>, 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(<>, left, count) do skip_names(rest, left - 1, count) end end defp decode_types(<<>>, 0, _types), do: [] defp decode_types(<>, 0, types) do types = types |> decode_types() |> :lists.reverse() decode_rows(types, rest, _row = [], _rows = [], types) end defp decode_types(<>, count, acc) do decode_types(rest, count - 1, [type | acc]) end @doc false def decode_types([type | types]) do [decode_type(type) | decode_types(types)] end def decode_types([] = done), do: done for {encoded, decoded} <- scalar_types do defp decode_type(<>), do: unquote(decoded) end defp decode_type("DateTime('" <> rest) do [timezone] = :binary.split(rest, ["'", ")"], [:global, :trim_all]) {:datetime, timezone} end defp decode_type("DateTime64(" <> rest) do case :binary.split(rest, [", ", ")", "'"], [:global, :trim_all]) do [precision, timezone] -> time_unit = round(:math.pow(10, String.to_integer(precision))) {:datetime64, time_unit, timezone} [precision] -> time_unit = round(:math.pow(10, String.to_integer(precision))) {:datetime64, time_unit, nil} end end defp decode_type("DateTime" <> _), do: {:datetime, _timezone = nil} defp decode_type("Date" <> _), do: :date defp decode_type("FixedString(" <> rest) do [size] = :binary.split(rest, ")", [:global, :trim]) string(size: String.to_integer(size)) end defp decode_type("Decimal(" <> rest) do [precision, scale] = :binary.split(rest, [", ", ")"], [:global, :trim]) {scale, _} = Integer.parse(scale) precision = String.to_integer(precision) decimal(size: decimal_size(precision), scale: scale) end defp decode_type("LowCardinality(" <> rest) do decode_type(rest) end defp decode_type("Array(" <> rest) do {:array, decode_type(rest)} end defp decode_type("Nullable(" <> rest) do {:nullable, decode_type(rest)} end defp decode_type("Enum8('" <> rest) do mapping = rest |> :binary.split(["' = ", ", '", ")"], [:global, :trim_all]) |> Enum.chunk_every(2) |> Map.new(fn [k, v] -> {String.to_integer(v), k} end) {:enum8, mapping} end defp decode_type("Enum16('" <> rest) do mapping = rest |> :binary.split(["' = ", ", '", ")"], [:global, :trim_all]) |> Enum.chunk_every(2) |> Map.new(fn [k, v] -> {String.to_integer(v), k} end) {:enum16, mapping} end defp decode_type(type) do raise ArgumentError, "#{type} type is not supported" end @compile inline: [decode_string_decode_rows: 5] for {pattern, size} <- varints do defp decode_string_decode_rows( <>, 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(<>, 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(<>, 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( <>, 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( <>, 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 defp decode_rows([type | types_rest], <>, row, rows, types) do case type do :u8 -> <> = bin decode_rows(types_rest, bin, [u | row], rows, types) :u16 -> <> = bin decode_rows(types_rest, bin, [u | row], rows, types) :u32 -> <> = bin decode_rows(types_rest, bin, [u | row], rows, types) :u64 -> <> = bin decode_rows(types_rest, bin, [u | row], rows, types) :u128 -> <> = bin decode_rows(types_rest, bin, [u | row], rows, types) :u256 -> <> = bin decode_rows(types_rest, bin, [u | row], rows, types) :i8 -> <> = bin decode_rows(types_rest, bin, [i | row], rows, types) :i16 -> <> = bin decode_rows(types_rest, bin, [i | row], rows, types) :i32 -> <> = bin decode_rows(types_rest, bin, [i | row], rows, types) :i64 -> <> = bin decode_rows(types_rest, bin, [i | row], rows, types) :i128 -> <> = bin decode_rows(types_rest, bin, [i | row], rows, types) :i256 -> <> = bin decode_rows(types_rest, bin, [i | row], rows, types) :f32 -> case bin do <> -> 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 <> -> 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? string(size: size) -> <> = 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 -> <> = bin uuid = <> decode_rows(types_rest, bin, [uuid | row], rows, types) :date -> <> = bin decode_rows(types_rest, bin, [Date.add(@epoch_date, d) | row], rows, types) :date32 -> <> = bin decode_rows(types_rest, bin, [Date.add(@epoch_date, d) | row], rows, types) {:datetime, timezone} -> <> = 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: size, scale: scale) -> <> = 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) {:datetime64, time_unit, timezone} -> <> = 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} -> <> = bin decode_rows(types_rest, bin, [Map.fetch!(mapping, v) | row], rows, types) {:enum16, mapping} -> <> = bin decode_rows(types_rest, bin, [Map.fetch!(mapping, v) | row], rows, types) :ipv4 -> <> = bin decode_rows(types_rest, bin, [{b1, b2, b3, b4} | row], rows, types) :ipv6 -> <> = bin decode_rows(types_rest, bin, [{b1, b2, b3, b4, b5, b6, b7, b8} | row], rows, types) end end defp decode_rows([], <<>>, row, rows, _types) do :lists.reverse([:lists.reverse(row) | rows]) end defp decode_rows([], <>, 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 end