Packages
ecto_ch
0.3.2
0.10.0
0.9.4
0.9.3
0.9.2
0.9.1
0.9.0
0.8.9
0.8.8
0.8.7
0.8.6
0.8.5
retired
0.8.4
0.8.3
0.8.2
0.8.1
0.8.0
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.6
0.6.5
0.6.4
0.6.3
0.6.2
retired
0.6.1
retired
0.6.0
retired
0.5.1
0.5.0
retired
0.4.1
0.4.0
retired
0.3.10
0.3.9
0.3.8
0.3.7
0.3.6
0.3.5
0.3.4
0.3.3
0.3.2
0.3.1
0.3.0
0.2.2
0.2.1
0.2.0
0.1.11
0.1.10
0.1.9
0.1.8
0.1.7
0.1.6
0.1.5
0.1.4
0.1.3
0.1.2
0.1.1
0.1.0
ClickHouse adapter for Ecto
Current section
Files
Jump to
Current section
Files
lib/ecto/adapters/clickhouse/connection.ex
defmodule Ecto.Adapters.ClickHouse.Connection do
@moduledoc false
@behaviour Ecto.Adapters.SQL.Connection
@dialyzer :no_improper_lists
alias Ecto.SubQuery
alias Ecto.Query.{QueryExpr, JoinExpr, BooleanExpr, WithExpr, Tagged}
@parent_as __MODULE__
@impl true
def child_spec(opts) do
Ch.child_spec(opts)
end
@impl true
def prepare_execute(conn, _name, statement, params, opts) do
query = Ch.Query.build(statement, opts[:command])
DBConnection.prepare_execute(conn, query, params, opts)
end
@impl true
def execute(conn, query, params, opts) do
DBConnection.execute(conn, query, params, opts)
end
# TODO what should be done about transactions? probably will need to build custom Repo.stream
@impl true
def query(conn, statement, params, opts) do
Ch.query(conn, statement, params, opts)
end
@impl true
def query_many(_conn, _statement, _params, _opts) do
raise "not implemented"
end
@impl true
def stream(conn, statement, params, opts) do
Ch.stream(conn, statement, params, opts)
end
@impl true
def to_constraints(_exception, _opts) do
raise "not implemented"
end
@impl true
def all(query, params \\ [], as_prefix \\ []) do
if Map.get(query, :lock) do
raise ArgumentError, "ClickHouse does not support locks"
end
sources = create_names(query, as_prefix)
cte = cte(query, sources, params)
from = from(query, sources, params)
select = select(query, sources, params)
join = join(query, sources, params)
where = where(query, sources, params)
group_by = group_by(query, sources, params)
having = having(query, sources, params)
window = window(query, sources, params)
order_by = order_by(query, sources, params)
limit = limit(query, sources, params)
offset = offset(query, sources, params)
combinations = combinations(query, params)
[
cte,
select,
from,
join,
where,
group_by,
having,
window,
order_by,
limit,
offset,
combinations
]
end
@dialyzer {:no_return, update_all: 1, update_all: 2}
@impl true
def update_all(query, _prefix \\ nil) do
raise Ecto.QueryError,
query: query,
message: "ClickHouse does not support UPDATE statements -- use ALTER TABLE instead"
end
@impl true
def delete_all(query, params \\ []) do
unless query.joins == [] do
raise Ecto.QueryError,
query: query,
message: "ClickHouse does not support JOIN on DELETE statements"
end
if query.select do
raise Ecto.QueryError,
query: query,
message: "ClickHouse does not support RETURNING on DELETE statements"
end
if query.with_ctes do
raise Ecto.QueryError,
query: query,
message: "ClickHouse does not support CTEs (WITH) on DELETE statements"
end
%{sources: sources} = query
{table, _schema, prefix} = elem(sources, 0)
where =
case query.wheres do
[] -> " WHERE 1"
_ -> where(query, {{nil, nil, nil}}, params)
end
["DELETE FROM ", quote_table(prefix, table) | where]
end
@impl true
def ddl_logs(_), do: []
@impl true
def table_exists_query(table) do
{"SELECT name FROM system.tables WHERE name={$0:String} LIMIT 1", [table]}
end
@impl true
def execute_ddl(command) do
Ecto.Adapters.ClickHouse.Migration.execute_ddl(command)
end
@impl true
def insert(prefix, table, header, rows, _on_conflict, returning, _placeholders) do
unless returning == [] do
raise ArgumentError, "ClickHouse does not support RETURNING on INSERT statements"
end
insert(prefix, table, header, rows)
end
def insert(prefix, table, header, rows) do
insert =
case header do
[] ->
["INSERT INTO " | quote_table(prefix, table)]
_not_empty ->
fields = [?(, intersperse_map(header, ?,, "e_name/1), ?)]
["INSERT INTO ", quote_table(prefix, table) | fields]
end
case rows do
{%Ecto.Query{} = query, params} -> [insert, ?\s | all(query, params)]
rows when is_list(rows) -> insert
end
end
@impl true
def update(_prefix, _table, _fields, _filters, _returning) do
raise ArgumentError,
"ClickHouse does not support UPDATE statements -- use ALTER TABLE instead"
end
@impl true
# https://clickhouse.com/docs/en/sql-reference/statements/delete
def delete(prefix, table, filters, returning) do
unless returning == [] do
raise ArgumentError, "ClickHouse does not support RETURNING on DELETE statements"
end
filters =
filters
|> Enum.with_index()
|> intersperse_map(" AND ", fn
{{field, nil}, _} ->
[quote_name(field), " IS NULL"]
{{field, value}, idx} ->
[quote_name(field), ?=, build_param(idx, param_type(value))]
end)
["DELETE FROM ", quote_table(prefix, table), " WHERE ", filters]
end
@impl true
def explain_query(conn, query, params, opts) do
explain =
case Keyword.get(opts, :type, :plan) do
:ast -> "EXPLAIN AST "
:syntax -> "EXPLAIN SYNTAX "
:query_tree -> "EXPLAIN QUERY TREE "
:plan -> "EXPLAIN PLAN "
:pipeline -> "EXPLAIN PIPELINE "
:table_override -> "EXPLAIN TABLE OVERRIDE"
end
explain_query = [explain | query]
with {:ok, %{rows: rows}} <- query(conn, explain_query, params, opts) do
{:ok, rows}
end
end
binary_ops = [
==: " = ",
!=: " != ",
<=: " <= ",
>=: " >= ",
<: " < ",
>: " > ",
+: " + ",
-: " - ",
*: " * ",
/: " / ",
and: " AND ",
or: " OR ",
ilike: " ILIKE ",
like: " LIKE ",
# TODO these two are not in binary_ops in sqlite3 adapter
in: " IN ",
is_nil: " WHERE "
]
@binary_ops Keyword.keys(binary_ops)
for {op, str} <- binary_ops do
defp handle_call(unquote(op), 2), do: {:binary_op, unquote(str)}
end
defp handle_call(fun, _arity), do: {:fun, Atom.to_string(fun)}
defp select(%{select: %{fields: fields}, distinct: distinct} = query, sources, params) do
[
"SELECT ",
distinct(distinct, sources, params, query)
| select_fields(fields, sources, params, query)
]
end
defp select_fields([], _sources, _params, _query), do: "true"
defp select_fields(fields, sources, params, query) do
intersperse_map(fields, ?,, fn
# this is useful in array joins lie
#
# "arrays_test"
# |> join(:array, [a], r in "arr")
# |> select([a, r], {a.s, r})
#
{:&, _, [idx]} ->
{_, source, _} = elem(sources, idx)
source
{k, v} ->
[expr(v, sources, params, query), " AS " | quote_name(k)]
v ->
expr(v, sources, params, query)
end)
end
defp distinct(nil, _sources, _params, _query), do: []
defp distinct(%{expr: true}, _sources, _params, _query), do: "DISTINCT "
defp distinct(%{expr: false}, _sources, _params, _query), do: []
defp distinct(%{expr: exprs}, sources, params, query) when is_list(exprs) do
[
"DISTINCT ON (",
intersperse_map(exprs, ?,, &order_by_expr(&1, sources, params, query)) | ") "
]
end
defp from(%{from: %{source: source, hints: hints}} = query, sources, params) do
{from, name} = get_source(query, sources, params, 0, source)
[" FROM ", from, " AS ", name | hints(hints)]
end
def cte(
%{with_ctes: %WithExpr{recursive: recursive, queries: [_ | _] = queries}} = query,
sources,
params
) do
recursive_opt = if recursive, do: "RECURSIVE ", else: ""
ctes =
intersperse_map(queries, ?,, fn {name, _opts, cte} ->
[quote_name(name), " AS ", cte_query(cte, sources, params, query)]
end)
["WITH ", recursive_opt, ctes, " "]
end
def cte(%{with_ctes: _}, _sources, _params), do: []
defp cte_query(%Ecto.Query{} = query, sources, params, parent_query) do
query = put_in(query.aliases[@parent_as], {parent_query, sources})
[?(, all(query, params, subquery_as_prefix(sources)), ?)]
end
defp cte_query(%QueryExpr{expr: expr}, sources, params, query) do
expr(expr, sources, params, query)
end
defp join(%{joins: []}, _sources, _params), do: []
defp join(%{joins: joins} = query, sources, params) do
Enum.map(joins, fn
%JoinExpr{qual: qual, ix: ix, source: source, on: %QueryExpr{expr: on_exrp}, hints: hints} ->
unless hints == [] do
raise Ecto.QueryError,
query: query,
message: "ClickHouse does not support hints on JOIN"
end
{join, name} = get_source(query, sources, params, ix, source)
[
join_qual(qual),
join,
" AS ",
name
| join_on(qual, on_exrp, sources, params, query)
]
end)
end
defp join_on(:cross, true, _sources, _params, _query), do: []
defp join_on(array, true, _sources, _params, _query)
when array in [:array, :left_array, :inner_lateral, :left_lateral] do
[]
end
defp join_on(_qual, expr, sources, params, query) do
[" ON " | expr(expr, sources, params, query)]
end
defp join_qual(:inner), do: " INNER JOIN "
defp join_qual(t) when t in [:inner_lateral, :array], do: " ARRAY JOIN "
defp join_qual(t) when t in [:left_lateral, :left_array], do: " LEFT ARRAY JOIN "
defp join_qual(:left), do: " LEFT OUTER JOIN "
defp join_qual(:right), do: " RIGHT OUTER JOIN "
defp join_qual(:full), do: " FULL OUTER JOIN "
defp join_qual(:cross), do: " CROSS JOIN "
defp where(%{wheres: wheres} = query, sources, params) do
boolean(" WHERE ", wheres, sources, params, query)
end
defp having(%{havings: havings} = query, sources, params) do
boolean(" HAVING ", havings, sources, params, query)
end
defp group_by(%{group_bys: []}, _sources, _params), do: []
defp group_by(%{group_bys: group_bys} = query, sources, params) do
[
" GROUP BY "
| intersperse_map(group_bys, ?,, fn %QueryExpr{expr: expr} ->
intersperse_map(expr, ?,, &expr(&1, sources, params, query))
end)
]
end
defp window(%{windows: []}, _sources, _params), do: []
defp window(%{windows: windows} = query, sources, params) do
[
" WINDOW "
| intersperse_map(windows, ?,, fn {name, %{expr: kw}} ->
[quote_name(name), " AS " | window_exprs(kw, sources, params, query)]
end)
]
end
defp window_exprs(kw, sources, params, query) do
[
?(,
intersperse_map(kw, ?\s, &window_expr(&1, sources, params, query)),
?)
]
end
defp window_expr({:partition_by, fields}, sources, params, query) do
["PARTITION BY " | intersperse_map(fields, ?,, &expr(&1, sources, params, query))]
end
defp window_expr({:order_by, fields}, sources, params, query) do
["ORDER BY " | intersperse_map(fields, ?,, &order_by_expr(&1, sources, params, query))]
end
defp window_expr({:frame, {:fragment, _, _} = fragment}, sources, params, query) do
expr(fragment, sources, params, query)
end
defp order_by(%{order_bys: []}, _sources, _params), do: []
defp order_by(%{order_bys: order_bys} = query, sources, params) do
[
" ORDER BY "
| intersperse_map(order_bys, ?,, fn %{expr: expr} ->
intersperse_map(expr, ?,, &order_by_expr(&1, sources, params, query))
end)
]
end
defp order_by_expr({dir, expr}, sources, params, query) do
str = expr(expr, sources, params, query)
case dir do
:asc ->
str
:desc ->
[str | " DESC"]
:asc_nulls_first ->
[str | " ASC NULLS FIRST"]
:desc_nulls_first ->
[str | " DESC NULLS FIRST"]
:asc_nulls_last ->
[str | " ASC NULLS LAST"]
:desc_nulls_last ->
[str | " DESC NULLS LAST"]
_ ->
raise Ecto.QueryError,
query: query,
message: "ClickHouse does not support #{dir} in ORDER BY"
end
end
defp limit(%{limit: nil}, _sources, _params), do: []
defp limit(%{limit: %{expr: expr}} = query, sources, params) do
[" LIMIT ", expr(expr, sources, params, query)]
end
defp offset(%{offset: nil}, _sources, _params), do: []
defp offset(%{offset: %{expr: expr}} = query, sources, params) do
[" OFFSET ", expr(expr, sources, params, query)]
end
defp combinations(%{combinations: combinations}, params) do
Enum.map(combinations, &combination(&1, params))
end
defp combination({:union, query}, params), do: [" UNION (", all(query, params), ?)]
defp combination({:union_all, query}, params), do: [" UNION ALL (", all(query, params), ?)]
defp combination({:except, query}, params), do: [" EXCEPT (", all(query, params), ?)]
defp combination({:intersect, query}, params), do: [" INTERSECT (", all(query, params), ?)]
defp combination({:except_all, query}, _params) do
raise Ecto.QueryError,
query: query,
message: "ClickHouse does not support EXCEPT ALL"
end
defp combination({:intersect_all, query}, _params) do
raise Ecto.QueryError,
query: query,
message: "ClickHouse does not support INTERSECT ALL"
end
defp hints([_ | _] = hints) do
[" " | intersperse_map(hints, ?\s, &hint/1)]
end
defp hints([]), do: []
defp hint(hint) when is_binary(hint), do: hint
# TODO remove once it's no longer used in plausible/analytics
# https://github.com/elixir-ecto/ecto/pull/4254
defp hint({k, v}) when is_atom(k) and is_integer(v) do
[Atom.to_string(k), ?\s, Integer.to_string(v)]
end
defp boolean(_name, [], _sources, _params, _query), do: []
defp boolean(name, [%{expr: expr, op: op} | exprs], sources, params, query) do
result =
Enum.reduce(exprs, {op, paren_expr(expr, sources, params, query)}, fn
%BooleanExpr{expr: expr, op: op}, {op, acc} ->
{op, [acc, operator_to_boolean(op) | paren_expr(expr, sources, params, query)]}
%BooleanExpr{expr: expr, op: op}, {_, acc} ->
{op, [?(, acc, ?), operator_to_boolean(op) | paren_expr(expr, sources, params, query)]}
end)
[name | elem(result, 1)]
end
defp operator_to_boolean(:and), do: " AND "
defp operator_to_boolean(:or), do: " OR "
defp parens_for_select([first_expr | _] = expression) do
if is_binary(first_expr) and String.match?(first_expr, ~r/^\s*select/i) do
[?(, expression, ?)]
else
expression
end
end
defp paren_expr(expr, sources, params, query) do
[?(, expr(expr, sources, params, query), ?)]
end
defp expr({_type, [literal]}, sources, params, query) do
expr(literal, sources, params, query)
end
defp expr({:^, [], [ix]}, _sources, params, _query) do
build_param(ix, param_type_at(params, ix))
end
defp expr({:^, [], [ix, len]}, _sources, params, _query) when len > 0 do
[?(, build_params(ix, len, params), ?)]
end
# using an empty array literal since empty tuples are not allowed in ClickHouse
defp expr({:^, [], [_, 0]}, _sources, _params, _query), do: "[]"
defp expr({{:., _, [{:&, _, [ix]}, field]}, _, []}, sources, _params, _query)
when is_atom(field) do
quote_qualified_name(field, sources, ix)
end
defp expr({{:., _, [{:parent_as, _, [as]}, field]}, _, []}, _sources, _params, query)
when is_atom(field) do
{ix, sources} = get_parent_sources_ix(query, as)
quote_qualified_name(field, sources, ix)
end
defp expr({:&, _, [ix]}, sources, _params, _query) do
{_, source, _} = elem(sources, ix)
source
end
defp expr({:&, _, [idx, fields, _counter]}, sources, _params, query) do
{_, name, schema} = elem(sources, idx)
if is_nil(schema) and is_nil(fields) do
raise Ecto.QueryError,
query: query,
message:
"ClickHouse requires a schema module when using selector " <>
"#{inspect(name)} but none was given. " <>
"Please specify a schema or specify exactly which fields from " <>
"#{inspect(name)} you desire"
end
intersperse_map(fields, ?,, &[name, ?. | quote_name(&1)])
end
defp expr({:in, _, [_left, []]}, _sources, _params, _query), do: "0"
defp expr({:in, _, [left, right]}, sources, params, query) when is_list(right) do
args = intersperse_map(right, ?,, &expr(&1, sources, params, query))
[expr(left, sources, params, query), " IN (", args, ?)]
end
defp expr({:in, _, [_, {:^, _, [_ix, 0]}]}, _sources, _params, _query), do: "0"
defp expr({:in, _, [left, right]}, sources, params, query) do
[expr(left, sources, params, query), " IN ", expr(right, sources, params, query)]
end
defp expr({:is_nil, _, [arg]}, sources, params, query) do
[expr(arg, sources, params, query) | " IS NULL"]
end
defp expr({:not, _, [expr]}, sources, params, query) do
["NOT (", expr(expr, sources, params, query), ?)]
end
defp expr({:filter, _, [agg, filter]}, sources, params, query) do
[
expr(agg, sources, params, query),
" FILTER (WHERE ",
expr(filter, sources, params, query),
?)
]
end
defp expr(%SubQuery{query: query}, sources, params, parent_query) do
query = put_in(query.aliases[@parent_as], {parent_query, sources})
[?(, all(query, params, subquery_as_prefix(sources)), ?)]
end
defp expr({:fragment, _, [kw]}, _sources, _params, query)
when is_list(kw) or tuple_size(kw) == 3 do
raise Ecto.QueryError,
query: query,
message: "ClickHouse adapter does not support keyword or interpolated fragments"
end
defp expr({:fragment, _, parts}, sources, params, query) do
parts
|> Enum.map(fn
{:raw, part} -> part
{:expr, expr} -> expr(expr, sources, params, query)
end)
|> parens_for_select()
end
defp expr({:literal, _, [literal]}, _sources, _params, _query) do
quote_name(literal)
end
defp expr({:selected_as, _, [name]}, _sources, _params, _query) do
quote_name(name)
end
defp expr({:over, _, [agg, name]}, sources, params, query) when is_atom(name) do
[expr(agg, sources, params, query), " OVER " | quote_name(name)]
end
defp expr({:over, _, [agg, kw]}, sources, params, query) do
[expr(agg, sources, params, query), " OVER " | window_exprs(kw, sources, params, query)]
end
defp expr({:{}, _, elems}, sources, params, query) do
[?(, intersperse_map(elems, ?,, &expr(&1, sources, params, query)), ?)]
end
defp expr({:count, _, []}, _sources, _params, _query), do: "count(*)"
defp expr({:datetime_add, _, [datetime, count, interval]}, sources, params, query) do
[
expr(datetime, sources, params, query),
" + ",
interval(count, interval, sources, params, query)
]
end
defp expr({:date_add, _, [date, count, interval]}, sources, params, query) do
[
"CAST(",
expr(date, sources, params, query),
" + ",
interval(count, interval, sources, params, query)
| " AS Date)"
]
end
defp expr({:json_extract_path, _, [expr, path]}, sources, params, query) do
path =
Enum.map(path, fn
bin when is_binary(bin) -> [?., escape_json_key(bin)]
int when is_integer(int) -> [?[, Integer.to_string(int), ?]]
end)
["JSON_QUERY(", expr(expr, sources, params, query), ", '$", path | "')"]
end
# TODO parens?
defp expr({:exists, _, [subquery]}, sources, params, query) do
["exists" | expr(subquery, sources, params, query)]
end
defp expr({fun, _, args}, sources, params, query) when is_atom(fun) and is_list(args) do
{modifier, args} =
case args do
[rest, :distinct] -> {"DISTINCT ", [rest]}
_ -> {[], args}
end
case handle_call(fun, length(args)) do
{:binary_op, op} ->
[left, right] = args
[
op_to_binary(left, sources, params, query),
op | op_to_binary(right, sources, params, query)
]
{:fun, fun} ->
[fun, ?(, modifier, intersperse_map(args, ?,, &expr(&1, sources, params, query)), ?)]
end
end
defp expr(list, sources, params, query) when is_list(list) do
[?[, intersperse_map(list, ?,, &expr(&1, sources, params, query)), ?]]
end
defp expr(%Decimal{} = decimal, _sources, _params, _query) do
Decimal.to_string(decimal, :normal)
end
defp expr(%Tagged{value: value, type: :any}, sources, params, query) do
expr(value, sources, params, query)
end
defp expr(%Tagged{value: value, type: type}, sources, params, query) do
["CAST(", expr(value, sources, params, query), " AS ", ecto_to_db(type, query), ?)]
end
defp expr(nil, _sources, _params, _query), do: "NULL"
# TODO "true" / "false"?
defp expr(true, _sources, _params, _query), do: "1"
defp expr(false, _sources, _params, _query), do: "0"
defp expr(literal, _sources, _params, _query) when is_binary(literal) do
[?', escape_string(literal), ?']
end
defp expr(literal, _sources, _params, _query) when is_integer(literal) do
Integer.to_string(literal)
end
defp expr(literal, _sources, _params, _query) when is_float(literal) do
Float.to_string(literal)
end
defp expr(expr, _sources, _params, query) do
raise Ecto.QueryError,
query: query,
message: "unsupported expression #{inspect(expr)}"
end
defp op_to_binary({op, _, [_, _]} = expr, sources, params, query) when op in @binary_ops do
paren_expr(expr, sources, params, query)
end
defp op_to_binary({:is_nil, _, [_]} = expr, sources, params, query) do
paren_expr(expr, sources, params, query)
end
defp op_to_binary(expr, sources, params, query) do
expr(expr, sources, params, query)
end
defp create_names(%{sources: sources}, as_prefix) do
sources |> create_names(0, tuple_size(sources), as_prefix) |> List.to_tuple()
end
defp create_names(sources, pos, limit, as_prefix) when pos < limit do
[create_name(sources, pos, as_prefix) | create_names(sources, pos + 1, limit, as_prefix)]
end
defp create_names(_sources, pos, pos, as_prefix), do: [as_prefix]
defp subquery_as_prefix(sources) do
[?s | :erlang.element(tuple_size(sources), sources)]
end
defp create_name(sources, pos, as_prefix) do
case elem(sources, pos) do
{:fragment, _, _} ->
{nil, as_prefix ++ [?f | Integer.to_string(pos)], nil}
{table, schema, prefix} ->
name = as_prefix ++ [create_alias(table) | Integer.to_string(pos)]
{quote_table(prefix, table), name, schema}
%SubQuery{} ->
{nil, as_prefix ++ [?s | Integer.to_string(pos)], nil}
end
end
defp create_alias(<<first, _rest::bytes>>)
when first in ?a..?z
when first in ?A..?Z do
<<first>>
end
defp create_alias(_), do: ?t
@doc false
def intersperse_map([elem], _separator, mapper), do: [mapper.(elem)]
def intersperse_map([elem | rest], separator, mapper) do
[mapper.(elem), separator | intersperse_map(rest, separator, mapper)]
end
def intersperse_map([], _separator, _mapper), do: []
@compile inline: [build_param: 2]
defp build_param(ix, type) do
["{$", Integer.to_string(ix), ?:, type, ?}]
end
@doc false
def build_params(ix, len, params) when len > 1 do
[build_param(ix, param_type_at(params, ix)), ?, | build_params(ix + 1, len - 1, params)]
end
def build_params(ix, _len = 1, params), do: build_param(ix, param_type_at(params, ix))
def build_params(_ix, _len = 0, _params), do: []
@doc false
def quote_name(name, quoter \\ ?")
def quote_name(nil, _), do: []
def quote_name(names, quoter) when is_list(names) do
names
|> Enum.reject(&is_nil/1)
|> intersperse_map(?., "e_name(&1, nil))
|> wrap_in(quoter)
end
def quote_name(name, quoter) when is_atom(name) do
name |> Atom.to_string() |> quote_name(quoter)
end
def quote_name(name, quoter) do
wrap_in(name, quoter)
end
defp quote_qualified_name(name, sources, ix) do
{_, source, _} = elem(sources, ix)
case source do
nil -> quote_name(name)
_other -> [source, ?. | quote_name(name)]
end
end
@doc false
def quote_table(prefix, name)
def quote_table(nil, name), do: quote_name(name)
def quote_table(prefix, name), do: [quote_name(prefix), ?., quote_name(name)]
defp wrap_in(value, nil), do: value
defp wrap_in(value, wrapper), do: [wrapper, value, wrapper]
@doc false
# TODO faster?
def escape_string(value) when is_binary(value) do
value
|> :binary.replace("'", "''", [:global])
|> :binary.replace("\\", "\\\\", [:global])
end
defp escape_json_key(value) when is_binary(value) do
value
|> escape_string()
|> :binary.replace("\"", "\\\"", [:global])
end
defp get_source(query, sources, params, ix, source) do
{expr, name, _schema} = elem(sources, ix)
{expr || expr(source, sources, params, query), name}
end
defp get_parent_sources_ix(query, as) do
case query.aliases[@parent_as] do
{%{aliases: %{^as => ix}}, sources} -> {ix, sources}
{%{} = parent, _sources} -> get_parent_sources_ix(parent, as)
end
end
# TODO quote?
defp interval(count, interval, _sources, _params, _query) when is_integer(count) do
["INTERVAL ", Integer.to_string(count), ?\s, interval]
end
defp interval(count, interval, _sources, _params, _query) when is_float(count) do
count = :erlang.float_to_binary(count, [:compact, decimals: 16])
["INTERVAL ", count, ?\s, interval]
end
# TODO typecast to ::numeric?
defp interval(count, interval, sources, params, query) do
[expr(count, sources, params, query), " * ", interval(1, interval, sources, params, query)]
end
# when ecto migrator queries for versions in schema_versions it uses type(version, :integer)
# so we need :integer to be the same as :bigint which is used for schema_versions table definition
# this is why :integer is Int64 and not Int32
defp ecto_to_db(:integer, _query), do: "Int64"
defp ecto_to_db(:binary, _query), do: "String"
defp ecto_to_db({:parameterized, Ch, type}, _query), do: Ch.Types.encode(type)
defp ecto_to_db({:array, type}, query), do: ["Array(", ecto_to_db(type, query), ?)]
defp ecto_to_db(type, _query) when type in [:uuid, :string, :date, :boolean] do
Ch.Types.encode(type)
end
defp ecto_to_db(type, query) do
raise Ecto.QueryError,
query: query,
message: "unknown or ambiguous (for ClickHouse) Ecto type #{inspect(type)}"
end
defp param_type_at(params, ix) do
value = Enum.at(params, ix)
param_type(value)
end
defp param_type(s) when is_binary(s), do: "String"
defp param_type(i) when is_integer(i) and i > 0x7FFFFFFFFFFFFFFF, do: "UInt64"
defp param_type(i) when is_integer(i), do: "Int64"
defp param_type(f) when is_float(f), do: "Float64"
defp param_type(b) when is_boolean(b), do: "Bool"
# TODO Date32
defp param_type(%NaiveDateTime{}), do: "DateTime"
defp param_type(%DateTime{microsecond: microsecond}) do
case microsecond do
{_val, precision} when precision > 0 ->
"DateTime64"
_ ->
"DateTime"
end
end
defp param_type(%Date{}), do: "Date"
defp param_type(%Decimal{exp: exp}) do
# TODO use sizes 128 and 256 as well if needed
scale = if exp < 0, do: abs(exp), else: 0
["Decimal64(", Integer.to_string(scale), ?)]
end
defp param_type([]), do: "Array(Nothing)"
# TODO check whole list
defp param_type([v | _]), do: ["Array(", param_type(v), ?)]
end