Packages

SQL provides state-of-the-art, high-performance SQL integration for Elixir, built to handle extreme concurrency with unmatched expressiveness and ergonomic query composition. Write safe, composable, parameterized queries directly, without translating to Ecto or any ORM.

Current section

Files

Jump to
sql lib sql.ex
Raw

lib/sql.ex

# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: 2025 DBVisor
defmodule SQL do
@moduledoc "README.md"
|> File.read!()
|> String.split("<!-- MDOC !-->")
|> Enum.fetch!(1)
@moduledoc since: "0.1.0"
# require Logger
alias SQL.Adapters.ANSI
defmacro __using__(opts) do
quote bind_quoted: [opts: opts] do
Application.ensure_all_started(:sql, :permanent)
@doc false
import SQL
pool = opts[:pool] || :default
{_, config} = Enum.find(Application.compile_env(:sql, :pools, []), fn {p, _opts} -> p == pool end)
config = Map.new(Keyword.merge([case: :lower, columns: [], adapter: config[:adapter] || ANSI, validate: fn _, _ -> true end], opts))
@external_resource Path.relative_to_cwd("sql.lock")
config = with true <- File.exists?(Path.relative_to_cwd("sql.lock")),
%{validate: validate, columns: columns} <- elem(Code.eval_file("sql.lock", File.cwd!()), 0) do
%{config | validate: validate, columns: columns}
else
_ ->
%{config | columns: :persistent_term.get({pool, :columns})}
end
Module.put_attribute(__MODULE__, :sql_config, config)
Module.put_attribute(__MODULE__, :sql_pool, pool)
end
end
defstruct [tokens: [], idx: 0, params: [], module: nil, id: nil, string: nil, inspect: nil, fn: nil, context: nil, name: nil, columns: [], c_len: 0, types: [], pool: :default, portal: nil, p_len: 0, timeout: nil, acc: nil, max_rows: 0]
defimpl Inspect, for: SQL do
def inspect(%{inspect: nil, tokens: tokens, context: context}, _opts) do
{:current_stacktrace, stack} = Process.info(self(), :current_stacktrace)
SQL.__inspect__(tokens, context, hd(stack))
end
def inspect(%{inspect: inspect}, _opts), do: inspect
end
defimpl String.Chars, for: SQL do
def to_string(sql), do: sql.string
end
@doc """
Returns a parameterized SQL.
## Examples
iex(1)> email = "john@example.com"
iex(2)> SQL.to_sql(~SQL"select id, email from users where email = {{email}}")
{"select id, email from users where email = ?", ["john@example.com"]}
"""
@doc since: "0.1.0"
def to_sql(sql), do: {sql.string, sql.params}
@doc """
Handles the sigil `~SQL` for SQL.
It returns a `%SQL{}` struct that can be transformed to a parameterized query.
## Examples
iex(1)> ~SQL"from users select id, email"
~SQL\"\"\"
select
id,
email
from
users
\"\"\"
"""
@doc since: "0.1.0"
defmacro sigil_SQL(left \\ [], right, modifiers) do
SQL.build(left, right, modifiers, __CALLER__)
end
@doc """
Perform transformation on the result set.
## Examples
iex(1)> SQL.map(~SQL"from users select id, email", &IO.inspect/1)
~SQL\"\"\"
select
id,
email
from
users
\"\"\"
"""
@doc since: "0.4.0"
defmacro map(sql, fun) do
SQL.build(sql, fun, __CALLER__)
end
@doc """
Perform a transaction.
## Examples
iex(1)> SQL.transaction, do: Enum.list(~SQL"from users select id, email")
"""
@doc since: "0.5.0"
defmacro transaction(do: block) do
key = __CALLER__.function
pool = Module.get_attribute(__CALLER__.module, :sql_pool, :default)
quote do
caller = self()
ref = make_ref()
case :persistent_term.get(unquote(key), SQL.transaction()) do
nil ->
conn = conn(unquote(pool))
Process.put(SQL.Transaction, {caller, conn})
send conn, {ref, caller, System.monotonic_time(), ~SQL[begin]}
receive do
{^ref, :begin} ->
try do
result = unquote(block)
send conn, {ref, caller, System.monotonic_time(), ~SQL[commit]}
Process.delete(SQL.Transaction)
result
rescue
e ->
send conn, {ref, caller, System.monotonic_time(), ~SQL[rollback]}
Process.delete(SQL.Transaction)
{:error, e}
end
end
{owner, conn} = state ->
id = "sp_#{:erlang.unique_integer([:positive])}"
send conn, {owner, {ref, caller, System.monotonic_time(), parse("savepoint #{id}")}}
receive do
{^ref, :begin} ->
try do
result = unquote(block)
send conn, {owner, {ref, caller, System.monotonic_time(), parse("release savepoint #{id}")}}
result
rescue
e ->
send conn, {owner, {ref, caller, System.monotonic_time(), parse("rollback to savepoint #{id}")}}
{:error, e}
end
end
end
end
end
@doc false
@doc since: "0.5.0"
defmacro begin(key \\ __CALLER__.function, pool \\ Module.get_attribute(__CALLER__.module, :sql_pool)) do
quote do
ref = make_ref()
owner = self()
conn = conn(unquote(pool))
state = {owner, conn}
:persistent_term.put({unquote(key), :ref}, ref)
:persistent_term.put(unquote(key), state)
:persistent_term.put({SQL.Conn, owner}, state)
Process.put(SQL.Transaction, {owner, conn})
send conn, {ref, owner, System.monotonic_time(), ~SQL[begin]}
receive do
{^ref, :begin} -> :ok
end
end
end
@doc false
@doc since: "0.5.0"
defmacro rollback(key \\ __CALLER__.function) do
quote do
{owner, conn} = :persistent_term.get(unquote(key))
send conn, {:persistent_term.get({unquote(key), :ref}), owner, System.monotonic_time(), ~SQL[rollback]}
:persistent_term.erase({SQL.Conn, owner})
:persistent_term.erase(unquote(key))
:persistent_term.erase({unquote(key), :ref})
Process.delete(SQL.Transaction)
:ok
end
end
@doc false
@doc since: "0.5.0"
defmacro commit(key \\ __CALLER__.function) do
quote do
{owner, conn} = :persistent_term.get(unquote(key))
send conn, {:persistent_term.get({unquote(key), :ref}), owner, System.monotonic_time(), ~SQL[commit]}
:persistent_term.erase({SQL.Conn, owner})
:persistent_term.erase(unquote(key))
:persistent_term.erase({unquote(key), :ref})
Process.delete(SQL.Transaction)
:ok
end
end
@doc """
Returns a lazy enumerable.
## Examples
iex(1)> SQL.transaction, do: ~SQL"from users select id, email" |> SQL.stream() |> Stream.run()
"""
@doc since: "0.5.0"
defmacro stream(sql, opts \\ [max_rows: 500]) do
max_rows = Keyword.fetch!(opts, :max_rows)
quote do
%{unquote(sql) | max_rows: unquote(max_rows)}
end
end
@doc false
@doc since: "0.1.0"
def parse(binary, binding \\ [], module \\ ANSI) do
{:ok, context, tokens} = SQL.Lexer.lex(binary)
{:ok, context, tokens} = SQL.Parser.parse(tokens, context)
{:ok, columns, c_len, types, params} = SQL.Parser.describe(tokens, [])
id = :erlang.phash2(tokens)
portal = "p_#{:erlang.unique_integer([:positive])}"
struct(SQL, id: id, name: "sql_#{id}", tokens: tokens, context: context, string: IO.iodata_to_binary(module.to_iodata(tokens, context)), types: types, params: eval(params, binding, __ENV__, []), columns: columns, c_len: c_len, portal: portal, p_len: byte_size(portal))
end
@doc false
def build(left, {:<<>>, _, _} = right, _modifiers, env) do
config = %{case: :lower, adapter: Application.get_env(:sql, :adapter, ANSI), validate: fn _, _ -> true end}
config = if env.module, do: Module.get_attribute(env.module, :sql_config, config), else: config
columns = Map.get(config, :columns, [])
sql = struct(SQL, module: env.module)
stack = if env.function do
{env.module, elem(env.function, 0), elem(env.function, 1), [file: Path.relative_to_cwd(env.file), line: env.line]}
else
{env.module, env.function, 0, [file: Path.relative_to_cwd(env.file), line: env.line]}
end
case build(left, right) do
{:static, data} ->
id = id(data)
{:ok, context, tokens} = SQL.Lexer.lex(data, env.file)
{:ok, context, tokens} = SQL.Parser.parse(tokens, %{context|validate: config.validate, module: config.adapter, case: config.case})
{:ok, columns, c_len, types, params} = SQL.Parser.describe(tokens, columns)
string = IO.iodata_to_binary(context.module.to_iodata(tokens, context))
inspect = __inspect__(tokens, context, stack)
portal = "p_#{:erlang.unique_integer([:positive])}"
sql = %{sql | name: "sql_#{id}", params: [], columns: columns, c_len: c_len, types: types, tokens: tokens, string: string, inspect: inspect, id: id, portal: portal, p_len: byte_size(portal)}
case context.binding do
0 -> Macro.escape(sql)
_ ->
quote bind_quoted: [params: params, sql: Macro.escape(sql), env: Macro.escape(env)] do
portal = "p_#{:erlang.unique_integer([:positive])}"
%{sql | params: eval(params, binding(), env, sql.params), portal: portal, p_len: byte_size(portal)}
end
end
{:dynamic, data} ->
id = id(data)
sql = %{sql | id: id, name: "sql_#{id}"}
quote bind_quoted: [columns: Macro.escape(columns), left: Macro.unpipe(left), right: right, file: env.file, data: data, sql: Macro.escape(sql), env: Macro.escape(env), config: Macro.escape(%{config| validate: nil}), stack: Macro.escape(stack)] do
{t,p} = Enum.reduce(left, {[], []}, fn
{[], 0}, acc -> acc
{v, 0}, {t, p} -> {t++v.tokens, p++v.params}
end)
{:ok, context, tokens} = tokens(right, file, sql.id)
tokens = t++tokens
context = %{context|validate: config.validate, module: config.adapter, format: :dynamic}
{string, inspect, columns, c_len, types, params} = plan(tokens, context, sql.id, stack, columns)
portal = "p_#{:erlang.unique_integer([:positive])}"
%{sql | tokens: tokens, params: eval(params, binding(), env, p), types: types, columns: columns, c_len: c_len, portal: portal, p_len: byte_size(portal), string: string, inspect: inspect}
end
end
end
@doc false
def eval([], _binding, _env, acc), do: acc
def eval([value|rest], binding, env, acc) do
eval(rest, binding, env, [eval(value, binding, env)|acc])
end
defp eval(value, binding, env) do
case is_tuple(value) do
true ->
{v, _, _} = Code.eval_quoted_with_env(value, binding, env)
v
false -> value
end
end
@doc false
def build(left, {tag, _, _} = right, _env) when tag in ~w[fn &]a do
{_type, data, acc2} = left
|> Macro.unpipe()
|> Enum.reduce({:static, [], []}, fn
{[], 0}, acc -> acc
{{_, _, []} = r, 0}, {_, l, right} -> {:dynamic, Macro.pipe(l, r, 0), right}
{{:sigil_SQL, _meta, [{:<<>>, _, _}, []]} = r, 0}, {type, l, right} -> {type, Macro.pipe(l, r, 0), right}
{{{:.,_,[{_,_,[:SQL]},:map]},_,[left]}, 0}, {type, acc, acc2} -> {type, acc, [left|acc2]}
end)
[r | rest] = Enum.reverse([right|acc2])
right = Enum.reduce(rest, r, fn r, {t, m, [{t2, m2, [vars, block]}]} -> {t, m, [{t2, m2, [vars, quote(do: unquote(r).(unquote(block)))]}]} end)
quote bind_quoted: [left: data, right: right] do
%{left | fn: right}
end
end
@doc false
def build(left, {:<<>>, _, right}) do
left
|> Macro.unpipe()
|> Enum.reduce({:static, right}, fn
{[], 0}, acc -> acc
{{:sigil_SQL, _meta, [{:<<>>, _, value}, []]}, 0}, {type, acc} -> {type, [value, ?\s, acc]}
{{_, _, _} = var, 0}, {_, acc} -> {:dynamic, [var, ?\s, acc]}
end)
|> case do
{:static, data} -> {:static, IO.iodata_to_binary(data)}
{:dynamic, data} -> {:dynamic, data}
end
end
@doc false
def id(data) do
case :persistent_term.get(data, nil) do
nil ->
id = :erlang.phash2(data)
:persistent_term.put(data, id)
id
id -> id
end
end
@doc false
def tokens(binary, file, id) do
key = {id, :lex}
case :persistent_term.get(key, nil) do
nil ->
result = SQL.Lexer.lex(binary, file)
:persistent_term.put(key, result)
result
result ->
result
end
end
@doc false
def plan(tokens, context, id, stack, columns) do
__plan__(tokens, context, {context.module, id, :plan}, stack, columns)
end
defp __plan__(tokens, context, key, stack, columns) do
case :persistent_term.get(key, nil) do
nil ->
{:ok, context, tokens} = SQL.Parser.parse(tokens, context)
{:ok, columns, c_len, types, params} = SQL.Parser.describe(tokens, columns)
format = {IO.iodata_to_binary(context.module.to_iodata(tokens, context)), __inspect__(tokens, context, stack), columns, c_len, types, params}
:persistent_term.put(key, format)
format
format ->
format
end
end
@error IO.ANSI.red()
@reset IO.ANSI.reset()
@doc false
def __inspect__(tokens, context, stack) do
inspect = IO.iodata_to_binary([@reset, "~SQL\"\"\""|[SQL.Format.to_iodata(tokens, context, 0, true)|~c"\n\"\"\""]])
case context.errors do
[] -> inspect
errors ->
{:current_stacktrace, [_|t]} = Process.info(self(), :current_stacktrace)
IO.warn([?\n,format_error(errors), IO.iodata_to_binary([@reset, " ~SQL\"\"\""|[SQL.Format.to_iodata(tokens, context, 1, true)|~c"\n \"\"\""]])], [stack|t])
inspect
end
end
@doc false
def format_error(errors), do: Enum.group_by(errors, &elem(&1, 2)) |> Enum.reduce([], fn
{k, [{:special, _, _}]}, acc -> [acc|[" the operator", @error,k,@reset, " is invalid, did you mean any of #{__suggest__(k)}\n"]]
{k, [{:special, _, _}|_]=v}, acc -> [acc|[" the operator ",@error,k,@reset," is mentioned #{length(v)} times but is invalid, did you mean any of #{__suggest__(k)}\n"]]
{k, [_]}, acc -> [acc|[" the relation ",@error,k,@reset," does not exist\n"]]
{k, v}, acc -> [acc|[" the relation ",@error,k,@reset," is mentioned #{length(v)} times but does not exist\n"]]
end)
@doc false
def reduce(%SQL{fn: nil} = sql, acc, fun) do
reduce(%{sql | fn: fn row, acc -> fun.(row, acc) end, acc: acc})
end
def reduce(sql, acc, fun) do
reduce(%{sql | fn: fn row, acc -> fun.(sql.fn.(row), acc) end, acc: acc})
end
defp reduce(sql) do
ref = make_ref()
self = self()
timestamp = System.monotonic_time()
entry = {ref, self, timestamp, sql}
case transaction() do
nil ->
conn = pick(sql.pool)
send(conn, entry)
result = recv(conn, ref, sql.timeout || 15_000)
# Logger.debug(sql: sql, time: System.convert_time_unit(System.monotonic_time()-timestamp, :native, :millisecond))
result
{owner, conn} ->
send(conn, {owner, entry})
result = recv(conn, ref, sql.timeout || 15_000)
# Logger.debug(sql: sql, time: System.convert_time_unit(System.monotonic_time()-timestamp, :native, :millisecond))
result
end
end
defp recv(conn, ref, timeout) do
receive do
{^ref, {:supended, _}=msg} -> msg
{^ref, {:halted, _}=msg} -> msg
{^ref, {:cont, _}} -> recv(conn, ref, timeout)
{^ref, {:done, _}=msg} -> msg
{^ref, {:error, %{message: message}}} ->
raise RuntimeError, message
after
timeout ->
send(conn, {ref, :cancel})
recv(conn, ref, timeout)
end
end
@doc false
def conn(%SQL{pool: pool}), do: pick(pool)
def conn(pool), do: pick(pool)
if Application.compile_env(:sql, :env) == :test do
defp __conn__() do
{:links, links} = Process.info(self(), :links)
{:parent, parent} = Process.info(self(), :parent)
[parent|links]
|> Kernel.++(Process.get(:"$callers", []))
|> Kernel.++(Process.get(:"$ancestors", []))
|> Enum.uniq()
|> Enum.find_value(&:persistent_term.get({SQL.Conn, &1}, nil))
end
defp conn() do
case Process.get(SQL.Conn) || __conn__() do
{_, conn} -> conn
conn -> conn
end
end
@doc false
def transaction() do
Process.get(SQL.Transaction) || __conn__()
end
else
defp conn() do
Process.get(SQL.Conn)
end
@doc false
def transaction() do
Process.get(SQL.Transaction)
end
end
defp pick(pool) do
case conn() do
nil ->
conns = :persistent_term.get(pool)
conn = elem(conns, :erlang.phash2({self(), :rand.uniform(1_000_000)}, tuple_size(conns)-1))
Process.put(SQL.Conn, conn)
conn
conn ->
conn
end
end
@doc false
def __suggest__(k), do: Enum.join(SQL.Lexer.suggest_operator(:erlang.iolist_to_binary(k)), ", ")
defimpl Enumerable, for: SQL do
def count(_enumerable), do: {:error, __MODULE__}
def member?(_enumerable, _element), do: {:error, __MODULE__}
def reduce(enumerable, acc, fun), do: SQL.reduce(enumerable, acc, fun)
def slice(_enumerable), do: {:error, __MODULE__}
end
end