Packages
ch
0.2.8
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.ex
defmodule Ch do
@moduledoc "Minimal HTTP ClickHouse client."
alias Ch.{Connection, Query, Result}
@type common_option ::
{:database, String.t()}
| {:username, String.t()}
| {:password, String.t()}
| {:settings, Keyword.t()}
| {:timeout, timeout}
@type start_option ::
common_option
| {:scheme, String.t()}
| {:hostname, String.t()}
| {:port, :inet.port_number()}
| {:transport_opts, :gen_tcp.connect_option()}
| DBConnection.start_option()
@doc """
Start the connection process and connect to ClickHouse.
## Options
* `:scheme` - HTTP scheme, defaults to `"http"`
* `:hostname` - server hostname, defaults to `"localhost"`
* `:port` - HTTP port, defaults to `8123`
* `:transport_opts` - options to be given to the transport being used. See `Mint.HTTP1.connect/4` for more info
* `:database` - Database, defaults to `"default"`
* `:username` - Username
* `:password` - User password
* `:settings` - Keyword list of ClickHouse settings
* `:timeout` - HTTP receive timeout in milliseconds
* `:transport_opts` - options to be given to the transport being used. See `Mint.HTTP1.connect/4` for more info
* [`DBConnection.start_option()`](https://hexdocs.pm/db_connection/DBConnection.html#t:start_option/0)
"""
@spec start_link([start_option]) :: GenServer.on_start()
def start_link(opts \\ []) do
DBConnection.start_link(Connection, opts)
end
@doc """
Returns a supervisor child specification for a DBConnection pool.
See `start_link/1` for supported options.
"""
@spec child_spec([start_option]) :: :supervisor.child_spec()
def child_spec(opts) do
DBConnection.child_spec(Connection, opts)
end
@type query_option ::
common_option
| {:command, Ch.Query.command()}
| {:headers, [{String.t(), String.t()}]}
| {:format, String.t()}
# TODO remove
| {:encode, boolean}
| {:decode, boolean}
| DBConnection.connection_option()
@doc """
Runs a query and returns the result as `{:ok, %Ch.Result{}}` or
`{:error, Exception.t()}` if there was a database error.
## Options
* `:database` - Database
* `:username` - Username
* `:password` - User password
* `:settings` - Keyword list of settings
* `:timeout` - Query request timeout
* `:command` - Command tag for the query
* `:headers` - Custom HTTP headers for the request
* `:format` - Custom response format for the request
* `:decode` - Whether to automatically decode the response
* [`DBConnection.connection_option()`](https://hexdocs.pm/db_connection/DBConnection.html#t:connection_option/0)
"""
@spec query(DBConnection.conn(), iodata, params, [query_option]) ::
{:ok, Result.t()} | {:error, Exception.t()}
when params: map | [term] | [row :: [term]] | iodata | Enumerable.t()
def query(conn, statement, params \\ [], opts \\ []) do
query = Query.build(statement, opts)
with {:ok, _query, result} <- DBConnection.execute(conn, query, params, opts) do
{:ok, result}
end
end
@doc """
Runs a query and returns the result or raises `Ch.Error` if
there was an error. See `query/4`.
"""
@spec query!(DBConnection.conn(), iodata, params, [query_option]) :: Result.t()
when params: map | [term] | [row :: [term]] | iodata | Enumerable.t()
def query!(conn, statement, params \\ [], opts \\ []) do
query = Query.build(statement, opts)
DBConnection.execute!(conn, query, params, opts)
end
@doc false
@spec stream(DBConnection.t(), iodata, map | [term], [query_option]) :: Ch.Stream.t()
def stream(conn, statement, params \\ [], opts \\ []) do
query = Query.build(statement, opts)
%Ch.Stream{conn: conn, query: query, params: params, opts: opts}
end
# TODO drop
@doc false
@spec run(DBConnection.conn(), (DBConnection.t() -> any), Keyword.t()) :: any
def run(conn, f, opts \\ []) when is_function(f, 1) do
DBConnection.run(conn, f, opts)
end
if Code.ensure_loaded?(Ecto.ParameterizedType) do
@behaviour Ecto.ParameterizedType
@impl true
def type(params), do: {:parameterized, {Ch, params}}
@impl true
def init(opts) do
clickhouse_type =
opts[:raw] || opts[:type] ||
raise ArgumentError, "keys :raw or :type not found in: #{inspect(opts)}"
Ch.Types.decode(clickhouse_type)
end
@impl true
def load(value, _loader, _params), do: {:ok, value}
@impl true
def dump(value, _dumper, _params), do: {:ok, value}
@impl true
def cast(value, :string = type), do: Ecto.Type.cast(type, value)
def cast(value, :boolean = type), do: Ecto.Type.cast(type, value)
def cast(value, :uuid), do: Ecto.Type.cast(Ecto.UUID, value)
def cast(value, :date = type), do: Ecto.Type.cast(type, value)
def cast(value, :date32), do: Ecto.Type.cast(:date, value)
def cast(value, :datetime), do: Ecto.Type.cast(:naive_datetime, value)
def cast(value, {:datetime, "UTC"}), do: Ecto.Type.cast(:utc_datetime, value)
def cast(value, {:datetime64, _p}), do: Ecto.Type.cast(:naive_datetime_usec, value)
def cast(value, {:datetime64, _p, "UTC"}), do: Ecto.Type.cast(:utc_datetime_usec, value)
def cast(value, {:fixed_string, _s}), do: Ecto.Type.cast(:string, value)
for size <- [8, 16, 32, 64, 128, 256] do
def cast(value, unquote(:"i#{size}")), do: Ecto.Type.cast(:integer, value)
def cast(value, unquote(:"u#{size}")), do: Ecto.Type.cast(:integer, value)
end
for size <- [32, 64] do
def cast(value, unquote(:"f#{size}")), do: Ecto.Type.cast(:float, value)
end
def cast(value, {:decimal = type, _p, _s}), do: Ecto.Type.cast(type, value)
for size <- [32, 64, 128, 256] do
def cast(value, {unquote(:"decimal#{size}"), _s}) do
Ecto.Type.cast(:decimal, value)
end
end
def cast(value, {:array, type}), do: Ecto.Type.cast({:array, type(type)}, value)
def cast(value, {:nullable, type}), do: cast(value, type)
def cast(value, {:low_cardinality, type}), do: cast(value, type)
def cast(value, {:simple_aggregate_function, _name, type}), do: cast(value, type)
def cast(value, :ring), do: Ecto.Type.cast({:array, type(:point)}, value)
def cast(value, :polygon), do: Ecto.Type.cast({:array, type(:ring)}, value)
def cast(value, :multipolygon), do: Ecto.Type.cast({:array, type(:polygon)}, value)
def cast(nil, _params), do: {:ok, nil}
def cast(value, {enum, mappings}) when enum in [:enum8, :enum16] do
result =
case value do
_ when is_integer(value) -> List.keyfind(mappings, value, 1, :error)
_ when is_binary(value) -> List.keyfind(mappings, value, 0, :error)
_ -> :error
end
case result do
{_, _} -> {:ok, value}
:error = e -> e
end
end
def cast(value, :ipv4) do
case value do
{a, b, c, d} when is_number(a) and is_number(b) and is_number(c) and is_number(d) ->
{:ok, value}
_ when is_binary(value) ->
with {:error = e, _reason} <- :inet.parse_ipv4_address(to_charlist(value)), do: e
_ when is_list(value) ->
with {:error = e, _reason} <- :inet.parse_ipv4_address(value), do: e
_ ->
:error
end
end
def cast(value, :ipv6) do
case value do
{a, s, d, f, g, h, j, k}
when is_number(a) and is_number(s) and is_number(d) and is_number(f) and
is_number(g) and is_number(h) and is_number(j) and is_number(k) ->
{:ok, value}
_ when is_binary(value) ->
with {:error = e, _reason} <- :inet.parse_ipv6_address(to_charlist(value)), do: e
_ when is_list(value) ->
with {:error = e, _reason} <- :inet.parse_ipv6_address(value), do: e
_ ->
:error
end
end
def cast(value, :point) do
case value do
{x, y} when is_number(x) and is_number(y) -> {:ok, value}
_ -> :error
end
end
def cast(value, {:tuple, types}), do: cast_tuple(types, value)
def cast(value, {:map, key_type, value_type}), do: cast_map(value, key_type, value_type)
defp cast_tuple(types, values) when is_tuple(values) do
cast_tuple(types, Tuple.to_list(values), [])
end
defp cast_tuple(types, values) when is_list(values) do
cast_tuple(types, values, [])
end
defp cast_tuple(_types, _values), do: :error
defp cast_tuple([type | types], [value | values], acc) do
case cast(value, type) do
{:ok, value} -> cast_tuple(types, values, [value | acc])
:error = e -> e
end
end
defp cast_tuple([], [], acc), do: {:ok, List.to_tuple(:lists.reverse(acc))}
defp cast_tuple(_types, _values, _acc), do: :error
defp cast_map(value, key_type, value_type) when is_map(value) do
cast_map(Map.to_list(value), key_type, value_type)
end
defp cast_map(value, key_type, value_type) when is_list(value) do
cast_map(value, key_type, value_type, [])
end
defp cast_map(_value, _key_type, _value_type), do: :error
defp cast_map([{key, value} | kvs], key_type, value_type, acc) do
with {:ok, key} <- cast(key, key_type),
{:ok, value} <- cast(value, value_type) do
cast_map(kvs, key_type, value_type, [{key, value} | acc])
end
end
defp cast_map([], _key_type, _value_type, acc), do: {:ok, Map.new(acc)}
defp cast_map(_kvs, _key_type, _value_type, _acc), do: :error
@impl true
def embed_as(_, _), do: :self
@impl true
def equal?(a, b, _), do: a == b
@impl true
def format(params) do
"#Ch<#{Ch.Types.encode(params)}>"
end
end
end