defmodule EctoQLC.Adapters.QLC do
  @moduledoc ~S"""
  This application provides functionality for working with Erlang databases in `Ecto`.

  ## Built-in adapters

    * `EctoQLC.Adapters.DETS` for [`dets`](https://www.erlang.org/doc/man/dets.html)
    * `EctoQLC.Adapters.ETS` for [`ets`](https://www.erlang.org/doc/man/ets.html)
    * `EctoQLC.Adapters.Mnesia` for [`mnesia`](https://www.erlang.org/doc/man/mnesia.html)

  ## Migrations

  `ecto_qlc` supports `ecto_sql` database migrations, currently none of the adapters support constraints, unique index or multi column index.
  """
  require Kernel
  require Logger
  alias Ecto.Migration.Table
  alias Ecto.Migration.Index
  alias Ecto.Migration.Constraint

  @doc false
  defmacro __using__(opts) do
    quote do
      @behaviour Ecto.Adapter
      @behaviour Ecto.Adapter.Migration
      @behaviour Ecto.Adapter.Structure
      @behaviour Ecto.Adapter.Queryable
      @behaviour Ecto.Adapter.Schema
      @behaviour Ecto.Adapter.Storage
      @behaviour Ecto.Adapter.Transaction

      alias __MODULE__

      opts = unquote(opts)
      @driver Keyword.fetch!(opts, :driver)

      @impl Ecto.Adapter
      def ensure_all_started(_config, _type) do
        case Application.ensure_all_started(@driver) do
          {:ok, []} -> {:ok, [@driver]}
          {:ok, [@driver]} -> {:ok, [@driver]}
          {:error, _reason} -> {:ok, [@driver]}
        end
      end

      @impl Ecto.Adapter
      def init(config) do
        EctoQLC.Adapters.QLC.init(config, @driver)
      end

      @impl Ecto.Adapter
      def checkout(%{pid: pid}, _opts, fun) do
        Process.put({__MODULE__, pid}, true)
        result = fun.()
        Process.delete({__MODULE__, pid})
        result
      end

      @impl Ecto.Adapter
      def checked_out?(%{pid: pid} = _adapter_meta) do
        Process.get({__MODULE__, pid}) != nil
      end

      @impl Ecto.Adapter
      defmacro __before_compile__(_env), do: :ok

      @impl Ecto.Adapter
      def loaders({:map, _}, type),   do: [&Ecto.Type.embedded_load(type, &1, :json)]
      def loaders(match, type) when match in ~w[binary_id embed_id]a, do: [Ecto.UUID, type]
      def loaders(_, type), do: [type]

      @impl Ecto.Adapter
      def dumpers({:map, _}, type),   do: [&Ecto.Type.embedded_load(type, &1, :json)]
      def dumpers({:array, {:array, value}}, _type), do: [{:in, value}]
      def dumpers(match, type) when match in ~w[binary_id embed_id]a, do: [type, Ecto.UUID]
      def dumpers(_, type), do: [type]

      @impl Ecto.Adapter.Queryable
      def prepare(operation, %Ecto.Query{} = query) do
        {:nocache, {operation, query}}
      end

      @impl Ecto.Adapter.Queryable
      def execute(adapter_meta, query_meta, {:nocache, query}, params, options) do
        EctoQLC.Adapters.QLC.execute(adapter_meta, query_meta, query, params, options)
      end

      @impl Ecto.Adapter.Queryable
      def stream(adapter_meta, query_meta, {:nocache, query}, params, options) do
        EctoQLC.Adapters.QLC.stream(adapter_meta, query_meta, query, params, options)
      end

      @impl Ecto.Adapter.Schema
      def delete(adapter_meta, schema_meta, filters, returning, options) do
        EctoQLC.Adapters.QLC.delete(@driver, adapter_meta, schema_meta, filters, returning, options)
      end

      @impl Ecto.Adapter.Schema
      def insert(adapter_meta, schema_meta, fields, on_conflict, returning, options) do
        EctoQLC.Adapters.QLC.insert(@driver, adapter_meta, schema_meta, fields, on_conflict, returning, options)
      end

      @impl Ecto.Adapter.Schema
      def insert_all(adapter_meta, schema_meta, header, list, on_conflict, returning, placeholders, options) do
        EctoQLC.Adapters.QLC.insert_all(@driver, adapter_meta, schema_meta, header, list, on_conflict, returning, placeholders, options)
      end

      @impl Ecto.Adapter.Schema
      def update(adapter_meta, schema_meta, fields, filters, returning, options) do
        EctoQLC.Adapters.QLC.update(@driver, adapter_meta, schema_meta, fields, filters, returning, options)
      end

      @impl Ecto.Adapter.Schema
      # TODO :id should be based on the table size, since restarting the system can give the same id
      def autogenerate(:id), do: System.unique_integer([:positive, :monotonic])
      def autogenerate(:embed_id), do: Ecto.UUID.generate()
      def autogenerate(:binary_id), do: Ecto.UUID.bingenerate()

      @impl Ecto.Adapter.Migration
      def execute_ddl(adapter_meta, command, options) do
        EctoQLC.Adapters.QLC.execute_ddl(adapter_meta, command, options)
      end

      @impl Ecto.Adapter.Migration
      def lock_for_migrations(adapter_meta, options, fun) do
        EctoQLC.Adapters.QLC.lock_for_migrations(adapter_meta, options, fun)
      end

      @impl Ecto.Adapter.Migration
      def supports_ddl_transaction?(), do: false

      @impl Ecto.Adapter.Transaction
      def transaction(adapter_meta, opts, fun) do
        Process.put({adapter_meta.pid, :transaction}, true)
        v = fun.()
        Process.put({adapter_meta.pid, :transaction}, false)
        {:ok, v}
      end

      @impl Ecto.Adapter.Transaction
      def in_transaction?(adapter_meta), do: Process.get({adapter_meta.pid, :transaction}, false)

      @impl Ecto.Adapter.Transaction
      def rollback(adapter_meta, %_schema{} = value) do
        if in_transaction?(adapter_meta) do
          throw(value)
        else
          raise "cannot call rollback outside of transaction"
        end
      end

      @impl Ecto.Adapter.Structure
      def dump_cmd(args, opts, config) do
        EctoQLC.Adapters.QLC.dump_cmd(args, opts, config, @driver)
      end

      @impl Ecto.Adapter.Structure
      def structure_dump(default, config) do
        EctoQLC.Adapters.QLC.structure_dump(default, config, @driver)
      end

      @impl Ecto.Adapter.Structure
      def structure_load(default, config) do
        EctoQLC.Adapters.QLC.structure_load(default, config, @driver)
      end

      defoverridable [prepare: 2, execute: 5, stream: 5, execute_ddl: 3, loaders: 2, dumpers: 2, checked_out?: 1, checkout: 3, autogenerate: 1, ensure_all_started: 2, __before_compile__: 1, lock_for_migrations: 3, supports_ddl_transaction?: 0, transaction: 3, in_transaction?: 1, rollback: 2]
    end
  end

  @doc false
  def dump_cmd(_args, _opts, _config, _driver) do
    {"not_implemnted", 1}
  end

  @doc false
  def structure_dump(_default, _config, _driver) do
    {:error, "not_implemnted"}
  end

  @doc false
  def structure_load(_default, _config, _driver) do
    {:error, "not_implemnted"}
  end

  @doc false
  def init(config, :mnesia = driver) do
    dir = '#{Keyword.fetch!(config, :dir)}'
    File.mkdir_p!(dir)
    Application.put_env(:mnesia, :dir, dir)
    log = Keyword.get(config, :log, :debug)
    stacktrace = Keyword.get(config, :stacktrace, nil)
    telemetry_prefix = Keyword.fetch!(config, :telemetry_prefix)
    telemetry = {config[:repo], log, telemetry_prefix ++ [:query]}
    {:ok, DynamicSupervisor.child_spec(strategy: :one_for_one, name: Module.concat([config[:repo], driver])), %{telemetry: telemetry, stacktrace: stacktrace, opts: config}}
  end

  def init(config, driver) do
    log = Keyword.get(config, :log, :debug)
    stacktrace = Keyword.get(config, :stacktrace, nil)
    telemetry_prefix = Keyword.fetch!(config, :telemetry_prefix)
    telemetry = {config[:repo], log, telemetry_prefix ++ [:query]}
    {:ok, DynamicSupervisor.child_spec(strategy: :one_for_one, name: Module.concat([config[:repo], driver])), %{telemetry: telemetry, stacktrace: stacktrace, opts: config}}
  end

  @aggregates ~w[avg count max min sum]a
  @operators ~w[or and > < >= <= == === != + - * /]a

  @doc false
  def execute(%{adapter: EctoQLC.Adapters.Mnesia} = adapter_meta, query_meta, {operator, query}, params, options) do
    ## TODO Improve error when database does not exists
    prepareed = prepare(adapter_meta, query_meta, query, params, options)
    qlc = to_qlc(operator, prepareed)
    :mnesia.transaction(fn ->
      if lock = elem(prepareed, 2).lock do
        :mnesia.lock(elem(lock, 0), elem(lock, 1))
      end
      query_handle = to_query_handle(operator, prepareed, qlc)
      {query_time, values} = :timer.tc(:qlc, :eval, [query_handle, []])
      {decode_time, value} = :timer.tc(__MODULE__, :select, [values, operator, prepareed])
      log(value, get_source(query.sources), qlc, query_time, decode_time, 0, 0, operator, adapter_meta.telemetry, params, query, options ++ adapter_meta.opts)
    end)
    |> elem(1)
  end
  def execute(adapter_meta, query_meta, {operator, query}, params, options) do
    prepareed = prepare(adapter_meta, query_meta, query, params, options)
    qlc = to_qlc(operator, prepareed)
    query_handle = to_query_handle(operator, prepareed, qlc)
    {query_time, values} = :timer.tc(:qlc, :eval, [query_handle, []])
    {decode_time, value} = :timer.tc(__MODULE__, :select, [values, operator, prepareed])
    log(value, get_source(query.sources), qlc, query_time, decode_time, 0, 0, operator, adapter_meta.telemetry, params, query, options ++ adapter_meta.opts)
  end

  @doc false
  def stream(adapter_meta, query_meta, {operator, query}, params, options) do
    key = :erlang.timestamp
    prepareed = prepare(adapter_meta, query_meta, query, params, options)
    qlc = to_qlc(operator, prepareed)
    query_handle = to_query_handle(operator, prepareed, qlc)
    Stream.resource(
      fn ->
      Process.put(key, :erlang.timestamp)
      :qlc.cursor(query_handle, elem(prepareed, 4))
      end,
      fn cursor ->
        case :qlc.next_answers(cursor, options[:max_rows] || 500) do
          []   -> {:halt, cursor}
          rows ->  {[select(rows, :all, prepareed)], cursor}
        end
      end,
      fn cursor ->
        result = :qlc.delete_cursor(cursor)
        query_time = :timer.now_diff(:erlang.timestamp, Process.get(key))
        log(result, get_source(query.sources), qlc, query_time, 0, 0, 0, operator, adapter_meta.telemetry, params, query, options ++ adapter_meta.opts)
      end
    )
  end

  @doc false
  def get_source({%Ecto.SubQuery{} = subquery}), do: get_source(subquery.query.sources)
  def get_source({source, _module_, _prefix}) when is_binary(source), do: source
  def get_source(source) when is_tuple(source), do: get_source(elem(source, 0))

  @creates [:create, :create_if_not_exists]
  @drops [:drop, :drop_if_exists]

  @doc false
  def execute_ddl(adapter_meta, {_command, %Constraint{}}, _options) do
    {:ok, [{:warn, "#{adapter_meta.adapter} adapter does not support CONSTRAINT commands", []}]}
  end
  def execute_ddl(%{adapter: EctoQLC.Adapters.DETS} = adapter_meta, {command, %Table{} = table, _columns}, options) when command in @creates do
    options = Keyword.merge(adapter_meta.opts, List.wrap(table.options) ++ options)
    table = to_table(adapter_meta, table.name, table.prefix, options)
    file = if dir = Application.get_env(:dets, :dir), do: Path.join(dir, "#{table}"), else: table
    options = Keyword.take(Keyword.merge([file: '#{file}'],  options), ~w[access auto_save estimated_no_objects file max_no_slots min_no_slots keypos ram_file repair type]a)
    case :dets.open_file(table, options) do
      {:ok, ^table}   ->
        Enum.map(:dets.all, &:dets.sync/1)
        {:ok, []}
      {:error, reason} -> {:ok, [{:warn, "#{inspect(:mnesia.error_description(reason))}", []}]}
    end
  end
  def execute_ddl(%{adapter: EctoQLC.Adapters.ETS} = adapter_meta, {command, %Table{} = table, _columns}, options) when command in @creates do
    options = Keyword.merge([write_concurrency: true, read_concurrency: true], Keyword.merge(adapter_meta.opts, List.wrap(table.options) ++ options))
    options = [:set, :public, :named_table, {:heir, GenServer.whereis(:__ecto_qlc__) || self(), %{}}] ++ Keyword.take(options, ~w[write_concurrency read_concurrency keypos heir heir decentralized_counters compressed]a)
    table = to_table(adapter_meta, table.name, table.prefix, options)
    with :undefined <- :ets.info(table),
         ^table     <- :ets.new(table, options) do
      {:ok, []}
        else
        _ ->
      {:ok, []}
    end
  end
  def execute_ddl(%{adapter: EctoQLC.Adapters.Mnesia} = adapter_meta, {command, %Table{} = table, columns}, options) when command in @creates do
    options = if "schema_migrations" == table.name do
      [disc_only_copies: [node() | Node.list]]
    else
      Keyword.take(Keyword.merge(adapter_meta.opts, List.wrap(table.options) ++ options), ~w[disc_copies access_mode disc_only_copies index load_order majority ram_copies record_name snmp storage_properties type local_content]a)
    end
    table = to_table(adapter_meta, table.name, table.prefix, options)
    primary_keys = Enum.count(columns, fn {_command, _column, _type, options} -> options[:primary_key] == true end)
    attributes = Enum.reject(columns, fn {_command, _column, _type, options} -> options[:primary_key] == true end) |> Enum.map(&elem(&1, 1))
    attributes = if primary_keys > 1, do: [:primary_keys | attributes], else: Enum.map(columns, &elem(&1, 1))

    with :ok <- :mnesia.start(),
         {:atomic, :ok} <- :mnesia.create_table(table, [{:attributes, attributes} | options]) do
          {:ok, []}
    else
      {:aborted, {:already_exists, ^table}} -> {:ok, []}

      {status, reason} when status in ~w[error aborted]a  -> {:ok, [{:warn, "#{inspect(:mnesia.error_description(reason))}", []}]}
    end
  end
  def execute_ddl(%{adapter: EctoQLC.Adapters.DETS} = adapter_meta, {command, %Table{} = table, _columns}, options) when command in @drops do
    table = to_table(adapter_meta, table.name, table.prefix, options)
    case :dets.close(table) do
      :ok -> {:ok, []}
      {:error, resoan} -> {:ok, [{:warn, "#{inspect(resoan)}", []}]}
    end
  end
  def execute_ddl(%{adapter: EctoQLC.Adapters.ETS} = adapter_meta, {command, %Table{} = table, _columns}, options) when command in @drops do
    table = to_table(adapter_meta, table.name, table.prefix, options)
    :ets.delete(table)
    {:ok, []}
  end
  def execute_ddl(%{adapter: EctoQLC.Adapters.Mnesia} = adapter_meta, {command, %Table{} = table, _columns}, options) when command in @drops do
    case :mnesia.delete_table(to_table(adapter_meta, table.name, table.prefix, options)) do
      {:atomic, :ok} -> {:ok, []}
      {:aborted, resoan} -> {:ok, [{:warn, "#{inspect(resoan)}", []}]}
    end
  end
  def execute_ddl(%{adapter: EctoQLC.Adapters.Mnesia} = adapter_meta, {:alter, %Table{} = table, changes}, options) do
    table = to_table(adapter_meta, table.name, table.prefix, options)
    attributes = :mnesia.table_info(table, :attributes)
    new_attributes = Enum.reduce(changes, attributes, fn change, attributes -> update_attributes(change, attributes) end)
    with true <- attributes != new_attributes,
      {:atomic, :ok} <- :mnesia.transform_table(table, &Enum.reduce(changes, &1, fn change, row -> update_row(adapter_meta, row, change, attributes) end), new_attributes) do
      {:ok, []}
    else
      false -> {:ok, []}
      {status, reason} when status in ~w[error aborted]a -> {:ok, [{:warn, "#{inspect(:mnesia.error_description(reason))}", []}]}
    end
  end
  def execute_ddl(adapter_meta, {:alter, %Table{} = _table, _changes}, _options) do
    {:ok, [{:warn, "#{adapter_meta.adapter} adapter does not support alter", []}]}
  end
  def execute_ddl(%{adapter: EctoQLC.Adapters.Mnesia} = adapter_meta, {command, %Index{columns: [column]} = index}, options) when command in @creates do
    case :mnesia.add_table_index(to_table(adapter_meta, index.name, index.prefix, options), column) do
      {:atomic, :ok} -> {:ok, []}
      {:aborted, {:already_exists, _table, _}} when command == :create_if_not_exists -> {:ok, []}
      {:aborted, {:already_exists, _table, _}} -> raise "index already exists"
      {:aborted, resoan} -> {:ok, [{:warn, "#{inspect(resoan)}", []}]}
    end
  end
  def execute_ddl(%{adapter: EctoQLC.Adapters.Mnesia} = adapter_meta, {command, %Index{columns: [column]} = index, _mode}, options) when command in @drops do
    case :mnesia.del_table_index(to_table(adapter_meta, index.name, index.prefix, options), column) do
      {:atomic, :ok} -> {:ok, []}
      {:aborted, {:no_exists, _table, _}} when command == :drop_if_exists -> {:ok, []}
      {:aborted, {:no_exists, _table, _}} -> raise "index does not exists"
      {:aborted, resoan} -> {:ok, [{:warn, "#{inspect(resoan)}", []}]}
    end
  end
  def execute_ddl(%{adapter: EctoQLC.Adapters.Mnesia}, {_command, %Index{}}, _options) do
    {:ok, [{:warn, "Mnesia adapter does not support index with multiply columns", []}]}
  end
  def execute_ddl(adapter_meta, {_command, %Index{}}, _options) do
    {:ok, [{:warn, "#{adapter_meta.adapter} adapter does not support index", []}]}
  end
  def execute_ddl(adapter_meta, {_command, %Index{}, _mode}, _options) do
    {:ok, [{:warn, "#{adapter_meta.adapter} adapter does not support index", []}]}
  end
  def execute_ddl(adapter_meta, {:rename, %Table{} = _current_table, %Table{} = _new_table}, _options) do
    # Since table name always stays the same then we need to copy the table into a newley created one.
    # There might be some limitations with also being aware of indexes
    # So for now we just gonna warn
    # current_table = to_table(adapter_meta, current_table.name, current_table.prefix, options)
    # new_table = to_table(adapter_meta, new_table.name, new_table.prefix, options)
    # case :mnesia.transform_table(current_table, &:erlang.setelement(1, &1, new_table), :mnesia.table_info(current_table, :attributes), new_table) do
    #   {:atomic, :ok} -> {:ok, []}
    #   {status, reason} when status in ~w[error aborted]a -> {:ok, [{:warn, "#{inspect(:mnesia.error_description(reason))}", []}]}
    # end
    {:ok, [{:warn, "#{adapter_meta.adapter} adapter does not support RENAME table commands", []}]}
  end
  def execute_ddl(%{adapter: EctoQLC.Adapters.Mnesia} = adapter_meta, {:rename, %Table{} = table, current_column, new_column}, options) do
    table = to_table(adapter_meta, table.name, table.prefix, options)
    attributes = Enum.map(:mnesia.table_info(table, :attributes), fn
      ^current_column -> new_column
      column -> column
    end)
    case :mnesia.transform_table(table, &(&1), attributes) do
      {:atomic, :ok} -> {:ok, []}
      {status, reason} when status in ~w[error aborted]a -> {:ok, [{:warn, "#{inspect(:mnesia.error_description(reason))}", []}]}
    end
  end
  def execute_ddl(adapter_meta, {:rename, %Table{} = _table, _current_column, _new_column}, _options) do
    {:ok, [{:warn, "#{adapter_meta.adapter} adapter does not support RENAME column commands", []}]}
  end
  def execute_ddl(adapter_meta, command, _options) when is_binary(command) do
    raise "#{adapter_meta.adapter} adapter does not support binary in execute"
  end
  def execute_ddl(adapter_meta, command, _options) when is_list(command) do
    raise "#{adapter_meta.adapter} adapter does not support keyword lists in execute"
  end

  @doc false
  def lock_for_migrations(%{adapter: EctoQLC.Adapters.Mnesia}, _options, fun) do
    with :ok   <- :global.sync(),
         :ok   <- :mnesia.start(),
         value when value != :aborted <- :global.trans({:lock_for_migrations, __MODULE__}, fun),
         v when v in [:ok, {:error, :no_such_log}] <- :mnesia.sync_log(),
         true  <- :global.del_lock({:lock_for_migrations, __MODULE__}) do
      value
    else
      reason -> {:error, reason}
    end
  end
  def lock_for_migrations(%{adapter: EctoQLC.Adapters.DETS}, _options, fun) do
    with :ok   <- :global.sync(),
      value when value != :aborted <- :global.trans({:lock_for_migrations, __MODULE__}, fun),
      _     <- Enum.map(:dets.all, &:dets.sync/1),
      true  <- :global.del_lock({:lock_for_migrations, __MODULE__}) do
      value
    else
      reason -> {:error, reason}
    end
  end
  def lock_for_migrations(_adapter_meta, _options, fun) do
    with :ok   <- :global.sync(),
      value when value != :aborted <- :global.trans({:lock_for_migrations, __MODULE__}, fun),
      true  <- :global.del_lock({:lock_for_migrations, __MODULE__}) do
      value
    else
      reason -> {:error, reason}
    end
  end

  defp update_attributes({command, column, _type, _options}, attributes) when command in [:remove_if_exists, :remove] do
    attributes -- [column]
  end
  defp update_attributes({command, column, _type, _options}, attributes) when command in [:add_if_not_exists, :add] do
    if column in attributes do
      attributes
    else
      attributes ++ [column]
    end
  end
  defp update_attributes(_change, attributes), do: attributes

  @doc false
  def update_row(_adapter_meta, row, {:add, _column, _type, options}, _attributes) do
    Tuple.append(row, options[:default])
  end
  def update_row(_adapter_meta, row, {:add_if_not_exists, column, _type, options}, attributes) do
    if column not in attributes, do: Tuple.append(row, options[:default]), else: row
  end
  def update_row(_adapter_meta, row, {:remove, column, _type, _options}, attributes) do
    Tuple.delete_at(row, Enum.find_index(attributes, &(&1 == column)))
  end
  def update_row(_adapter_meta, row, {:remove_if_exists, column, _type, _options}, attributes) do
    if column in attributes, do: Tuple.delete_at(row, Enum.find_index(attributes, &(&1 == column))), else: row
  end
  def update_row(_adapter_meta, row, {:modify, column, type, options}, attributes) do
    idx = Enum.find_index(attributes, &(&1 == column))
    :erlang.setelement(idx, row, cast(elem(row, idx), options[:from], type))
  end
  def update_row(adapter_meta, schema, fields, row) do
    Enum.reduce(fields, row, fn {column, value}, row ->
      idx = get_index(adapter_meta, column, schema.__schema__(:fields), schema.__schema__(:primary_key))
      :erlang.setelement(idx, row, value)
    end)
  end

  defp get_key([], [primary_key | _ ], fields), do: fields[primary_key]
  defp get_key([primary_key], _columns, fields), do: fields[primary_key]
  defp get_key(primary_keys, _columns, fields), do: Enum.reduce(primary_keys, {}, &Tuple.insert_at(&2, tuple_size(&2), fields[&1]))

  defp cast(value, _from, :list), do: '#{value}'
  defp cast(value, _from, :string), do: "#{value}"
  defp cast(value, _from, :integer) when is_binary(value) or is_list(value) do
    case Integer.parse("#{value}") do
      {integer, ""} -> integer
      result -> raise "Could no parse #{value} to integer got: #{inspect(result)}"
    end
  end
  defp cast(value, _from, :float) when is_binary(value) or is_list(value) do
    case Float.parse("#{value}") do
      {float, ""} -> float
      result -> raise "Could no parse #{value} to float got: #{inspect(result)}"
    end
  end

  defp bindings(params, bindings \\ :erl_eval.new_bindings()) do
    params
    |> Enum.reduce({length(bindings), bindings}, fn v, {count, bindings} ->
      count = count + 1
      {count, :erl_eval.add_binding(:"PARAM#{count}", v, bindings)}
    end)
    |> elem(1)
  end

  @doc false
  def coalesce(nil, nil), do: nil
  def coalesce(nil, right), do: right
  def coalesce(left, nil), do: left
  def coalesce(left, _), do: left

  @doc false
  def like(left, right) do
    String.match?(left, Regex.compile!(right))
  end

  @doc false
  def ilike(left, right) do
    String.match?(left, Regex.compile!(right, [:caseless]))
  end

  @doc false
  def to_match_spec(adapter_meta, schema, filters) do
    primary_key = schema.__schema__(:primary_key)
    columns = schema.__schema__(:fields) -- primary_key
    key = if length(primary_key) > 1, do: Enum.reduce(primary_key, {}, &Tuple.insert_at(&2, tuple_size(&2), filters[&1])), else: filters[hd(primary_key)]
    row = if adapter_meta.adapter == EctoQLC.Adapters.Mnesia, do: {to_table(adapter_meta, schema.__schema__(:source), schema.__schema__(:prefix), []), key}, else: {key}
    head = Enum.reduce(columns, row, fn column, head -> Tuple.insert_at(head, tuple_size(head), filters[column] || :"$#{tuple_size(head)}") end)
    body = if adapter_meta.adapter == EctoQLC.Adapters.Mnesia, do: [{head}], else: [true]
    conditions = []
    match_spec = [{head, conditions, body}]
    case :ets.test_ms(head, match_spec) do
      {:error, reason} -> raise RuntimeError, "invalid MatchSpec: #{inspect reason}"
      _ -> match_spec
    end
  end

  @doc false
  def prepare(%{adapter: adapter}, _query_meta, %Ecto.Query{lock: lock} = query, _params, _options) when not is_nil(lock) and adapter != EctoQLC.Adapters.Mnesia do
    raise Ecto.QueryError, query: query, message: "#{List.last(Module.split(adapter))} adapter does not support locks"
  end
  def prepare(_adapter_meta, _query_meta, %Ecto.Query{with_ctes: with_ctes} = query, _params, _options) when not is_nil(with_ctes) do
    raise Ecto.QueryError, query: query, message: "QLC adapter does not support CTE"
  end
  def prepare(_adapter_meta, _query_meta, %Ecto.Query{windows: windows} = query, _params, _options) when windows != [] do
    raise Ecto.QueryError, query: query, message: "QLC adapter does not support windows"
  end
  def prepare(_adapter_meta, _query_meta, %Ecto.Query{combinations: combinations} = query, _params, _options) when combinations != [] do
    raise Ecto.QueryError, query: query, message: "QLC adapter does not support combinations like: #{Enum.map_join(combinations, ", ", fn {k, _} -> k end)}"
  end
  def prepare(adapter_meta, query_meta, %Ecto.Query{} = query, params, options) do
    if query.select && Enum.any?(query.select.fields, &has_fragment/1), do: raise(Ecto.QueryError, query: query, message: "QLC adapter does not support fragemnt in select clauses")
    if query.wheres |> Enum.flat_map(&(&1.subqueries)) |> Enum.any?(&has_parent_as/1), do: raise(Ecto.QueryError, query: query, message: "QLC adapter does not support parent_as in a subquery's where clauses")
    options = options(query, options)
    prefix = options[:prefix] || query.from.prefix || query.prefix
    order_bys = if query.distinct && Keyword.keyword?(query.distinct.expr), do: [query.distinct | query.order_bys], else: query.order_bys
    {adapter_meta, query_meta, %{query |
      order_bys: order_bys,
      group_bys: group_bys(query),
      updates: updates(adapter_meta, query, params),
      offset: offset(query, params),
      lock: lock(adapter_meta, query, prefix),
      limit: limit(adapter_meta, query, prefix, params),
     }, params, options}
  end

  defp lock(_adapter_meta, %Ecto.Query{lock: nil}, _prefix), do: nil
  defp lock(adapter_meta, %Ecto.Query{lock: "write", from: %{source: {source, _module}}}, prefix), do: {{:table, to_table(adapter_meta, source, prefix, [])}, :write}
  defp lock(adapter_meta, %Ecto.Query{lock: "read", from: %{source: {source, _module}}}, prefix), do: {{:table, to_table(adapter_meta, source, prefix, [])}, :read}
  defp lock(adapter_meta, %Ecto.Query{lock: "sticky_write", from: %{source: {source, _module}}}, prefix), do: {{:table, to_table(adapter_meta, source, prefix, [])}, :sticky_write}
  defp lock(_adapter_meta, %Ecto.Query{lock: lock} = query, _prefix), do: raise(Ecto.QueryError, query: query, message: "Unsupported lock: #{inspect lock}, supported locks: write, read, stickey_write")

  defp offset(%Ecto.Query{offset: %{expr: {:^, _, [idx]}}}, params), do: Enum.at(params, idx, idx)
  defp offset(%Ecto.Query{offset: %{expr: expr}}, _params), do: expr
  defp offset(%Ecto.Query{}, _params), do: 0

  defp limit(adapter_meta, %Ecto.Query{limit: nil, from: %{source: {source, _module}}}, prefix, _params) do
    mod = :"#{String.downcase(List.last(Module.split(adapter_meta.adapter)))}"
    fun = if mod == :mnesia, do: :table_info, else: :info
    case apply(mod, fun, [to_table(adapter_meta, source, prefix, adapter_meta.opts), :size]) do
      :undefined -> 1
      0 -> 500
      limit -> limit
    end
  end
  defp limit(_adapter_meta, %Ecto.Query{limit: %{expr: {:^, _, [idx]}}}, _prefix, params), do: Enum.at(params, idx, idx)
  defp limit(_adapter_meta, %Ecto.Query{limit: %{expr: expr}}, _prefix, _params), do: expr
  defp limit(_adapter_meta, %Ecto.Query{limit: limit}, _prefix, _params), do: limit || 500

  defp group_bys(%Ecto.Query{group_bys: group_bys, sources: sources}) do
    Enum.reduce(group_bys, [], fn
      %Ecto.Query.QueryExpr{expr: [{:selected_as, [], [:date]}]}, acc -> acc

      %Ecto.Query.QueryExpr{expr: expr}, acc ->
        acc ++  for {{:., _, [{:&, _, [idx]}, column]}, _, _} <- expr do
          module = elem(elem(sources, idx), 1)
          {column, get_index(%{adapter: nil}, column, module.__schema__(:fields), module.__schema__(:primary_key))}
        end
    end)
  end

  defp updates(_adapter_meta, %Ecto.Query{updates: [] = updates}, _params), do: updates
  defp updates(adapter_meta, %Ecto.Query{from: %{source: {_source, module}}, updates: updates}, params) do
    addition = if adapter_meta.adapter == EctoQLC.Adapters.Mnesia, do: 2, else: 1
    columns = module.__schema__(:fields)
    Enum.flat_map(updates, fn %Ecto.Query.QueryExpr{expr: [set: set]} ->
      Enum.map(set, fn
        {column, {:^, _, [idx]}} -> {column, {Enum.find_index(columns, &(&1 == column)) + addition, Enum.at(params, idx)}}
        {column, value} -> {column, {Enum.find_index(columns, &(&1 == column)) + addition, value}}
      end)
    end)
  end

  # Subqurries are currently not allowed to have parent_as due to having to plan which query to execute first, an example would be if the subquery would have to match on the FK from the main query or to evaluate columns from the main query.
  # in that case we would have to execute the main query before we could evaluate the subquery for then filtering the main query based on the result of the subquery
  defp has_parent_as(fields, acc \\ false)
  defp has_parent_as(%Ecto.SubQuery{query: query}, acc) do
    has_parent_as(query.select.fields, acc) || if Enum.find(query.wheres, &has_parent_as(&1.expr, acc)), do: true, else: false
  end
  defp has_parent_as({_op, _meta, children}, acc), do: has_parent_as(children, acc)
  defp has_parent_as(nil, acc), do: acc
  defp has_parent_as([], acc), do: acc
  defp has_parent_as([{{:., _, [{:parent_as, _, _}, _]}, _, _} | _fields], _acc), do: true
  defp has_parent_as([{_op, _meta, children} | fields], acc), do: has_parent_as(children, acc) || has_parent_as(fields, acc)
  defp has_parent_as([_ | fields], acc), do: has_parent_as(fields, acc)

  defp has_fragment(fields, acc \\ false)
  defp has_fragment(%Ecto.SubQuery{query: query}, acc), do: has_fragment(query.select.fields, acc)
  defp has_fragment(%Ecto.Query.Tagged{value: value}, acc), do: has_fragment(value, acc)
  defp has_fragment(nil, acc), do: acc
  defp has_fragment([], acc), do: acc
  defp has_fragment({:fragment, _meta, _children}, _acc), do: true
  defp has_fragment([{{:., _, [{:fragment, _, _}, _]}, _, _} | _fields], _acc), do: true
  defp has_fragment({_op, _meta, children}, acc), do: has_fragment(children, acc)
  defp has_fragment({_op, children}, acc), do: has_fragment(children, acc)
  defp has_fragment([{:fragment, _meta, _children} | _fields], _acc), do: true
  defp has_fragment([{_op, _meta, children} | fields], acc), do: has_fragment(children, acc) || has_fragment(fields, acc)
  defp has_fragment([_ | fields], acc), do: has_fragment(fields, acc)

  defp has_aggregates(fields, acc \\ false)
  defp has_aggregates(nil, acc), do: acc
  defp has_aggregates([], acc), do: acc
  defp has_aggregates([{_, {op, _meta, _children}} | _fields], false) when op in @aggregates, do: true
  defp has_aggregates([{op, _meta, _children} | _fields], false) when op in @aggregates, do: true
  defp has_aggregates([{_op, _meta, children} | fields], _acc) do
    if has_aggregates(children), do: true, else: has_aggregates(fields)
  end
  defp has_aggregates([_ | fields], acc), do: has_aggregates(fields, acc)

  defp options(%Ecto.Query{} = query, options) do
    unique = unique?(query)
    if options[:unique] && unique, do: raise(Ecto.QueryError, query: query, message: "QLC does not support mixing distinct in queries and unique options")
    options
    # |> Keyword.put_new(:unique, unique)
    |> Enum.take_while(fn
      {k, _v} -> k in ~w[max_lookup cache join lookup unique]a
      k -> k in ~w[cache unique]a
    end)
    |> Enum.map(fn
      {:join, join} when join not in ~w[any merge lookup nested_loop]a -> raise(Ecto.QueryError, query: query, message: "QLC only supports: :any, :merge, :lookup or :nested_loop joins, got: `#{inspect(join)}`")
      x -> x
    end)
  end

  defp unique?(%Ecto.Query{distinct: %Ecto.Query.QueryExpr{}}), do: false
  defp unique?(%Ecto.Query{}), do: false


  defp to_qlc(:subquery = operator, query), do: '[#{to_expression(operator, query)} || #{to_qualifiers(query)}]'
  defp to_qlc(operator, query), do: '[#{to_expression(operator, query)} || #{to_qualifiers(query)}].'

  defp to_expression(operator, {adapter_meta, _query_meta, query, _params, options} = q) do
    mod = :"#{String.downcase(List.last(Module.split(adapter_meta.adapter)))}"
    count = tuple_size(query.sources) - 1
    if operator in ~w[delete_all update_all]a do
      if query.select do
        '{#{Enum.map_join(query.select.fields, ", ", &expr(&1, q))}}'
      else
        '#{Enum.map_join(0..count, ", ", fn idx -> "#{String.upcase(String.first(elem(elem(query.sources, idx), 0)))}#{idx}"  end)}'
      end
     else
      '{#{Enum.map_join(0..count, ", ", fn idx ->
        case elem(query.sources, idx) do
          {<<s::binary-size(1), _::binary>> = source, nil, prefix} ->
            table = to_table(adapter_meta, source, prefix, options)
            [primary_keys | fields] = if mod == :mnesia and table in :mnesia.system_info(:tables), do: :mnesia.table_info(table, :attributes), else: [:version, :inserted_at]
            primary_keys = if source == "schema_migrations", do: [:version], else: [primary_keys]
            Enum.map_join(fields, ", ", &to_element(adapter_meta, &1, fields, primary_keys, "#{String.upcase(s)}#{idx}"))

          {<<s::binary-size(1), _::binary>>, module, _} ->
            Enum.map_join(module.__schema__(:fields), ", ", &to_element(adapter_meta, &1, module.__schema__(:fields), module.__schema__(:primary_key), "#{String.upcase(s)}#{idx}"))

          %{query: %{sources: {{<<s::binary-size(1), _::binary>>, module, _}}}} ->
            Enum.map_join(module.__schema__(:fields), ", ", &to_element(adapter_meta, &1, module.__schema__(:fields), module.__schema__(:primary_key), "#{String.upcase(s)}#{idx}"))

          %{query: query} ->
            {<<s::binary-size(1), _::binary>>, module, _} = elem(query.sources, idx)
            Enum.map_join(module.__schema__(:fields), ", ", &to_element(adapter_meta, &1, module.__schema__(:fields), module.__schema__(:primary_key), "#{String.upcase(s)}#{idx}"))

        end
        end)}}'
     end
   end

  defp to_qualifiers({adapter_meta, _query_meta, query, _params, options} = q) do
    prefix = options[:prefix] || query.from.prefix || query.prefix
    options = options(query, options)
    count = tuple_size(query.sources) - 1
    mod = :"#{String.downcase(List.last(Module.split(adapter_meta.adapter)))}"
    options = Keyword.merge(options, [n_objects: query.limit])
    take = if mod == :mnesia, do: ~w[lock traverse n_objects]a, else: ~w[traverse n_objects]a
    table_opts = Keyword.take(options, take)
    generators = Enum.map_join(0..count, ", ", fn idx ->
      case elem(query.sources, idx) do
      {<<s::binary-size(1), _::binary>> = source, _module, _} ->
      "#{String.upcase(s)}#{idx} <- #{mod}:table('#{to_table(adapter_meta, source, prefix, options)}', [#{Enum.map_join(table_opts, ", ", fn {k, v} -> "{#{k}, #{v}}" end)}])"
      %{query: %{sources: {{<<s::binary-size(1), _::binary>> = source, _module, _}}}} ->
        "#{String.upcase(s)}#{idx} <- #{mod}:table('#{to_table(adapter_meta, source, prefix, options)}', [#{Enum.map_join(table_opts, ", ", fn {k, v} -> "{#{k}, #{v}}" end)}])"

      %{query: %{sources: sources}} ->
        {<<s::binary-size(1), _::binary>> = source, _module, _} = elem(sources, idx)
        "#{String.upcase(s)}#{idx} <- #{mod}:table('#{to_table(adapter_meta, source, prefix, options)}', [#{Enum.map_join(table_opts, ", ", fn {k, v} -> "{#{k}, #{v}}" end)}])"
      end
    end)
    filters = Enum.map_join(query.joins, " andalso ", fn %Ecto.Query.JoinExpr{prefix: _prefix, on: %Ecto.Query.QueryExpr{expr: expr}} -> expr(expr, q) end)
    guards = wheres(query.wheres, q, [])
    cond do
      filters == "" and guards == "" -> generators
      guards == "" -> "#{generators}, #{filters}"
      filters == "" -> "#{generators}, #{guards}"
      true -> "#{generators}, #{filters}, #{guards}"
    end
  end

  defp wheres([], _query, [" andalso "]), do: ""
  defp wheres([], _query, acc), do: to_string(acc)
  defp wheres([%Ecto.Query.BooleanExpr{expr: expr} | rest], query, [] = acc), do: wheres(rest, query, [expr(expr, query) | acc])
  defp wheres([%Ecto.Query.BooleanExpr{op: op, expr: expr} | rest], query, acc), do: wheres(rest, query, [acc] ++ "#{to_erlang_term(op)} #{expr(expr, query)}")

  defp to_query_handle(_operator, {_adapter_meta, _query_meta, query, params, options}, qlc) do
    Enum.reduce(query.order_bys, :qlc.string_to_handle(qlc, options, bindings(params)), fn %{expr: expr}, qh ->
      Enum.reduce(expr, qh, fn {k, {{:., _, [{:&, _, [idx]}, column]}, _, _}}, qh when k in ~w[asc desc]a  ->
        module = elem(elem(query.sources, idx), 1)
        columns = module.__schema__(:fields)
        primary_key = module.__schema__(:primary_key)

        key = case primary_key do
        [_primary_key] ->
          Enum.find_index(columns, &(&1 == column)) + 1
        primary_keys ->
          if idx = Enum.find_index(columns -- primary_keys, &(&1 == column)), do: idx + length(primary_keys), else: 1
        end
        :qlc.keysort(key, qh, order: to_order(query, k))

        {k, _v}, qh ->
          :qlc.sort(qh, order: to_order(query, k))
      end)
     end)
  end

  defp to_order(_query, :asc), do: :ascending
  defp to_order(_query, :desc), do: :descending
  defp to_order(query, order), do: raise(Ecto.QueryError, query: query, message: "QLC does not support ordering by: #{inspect order}")

  defp expr({_, _,[{{_, _, [{:parent_as, _, _}, _]}, _, _}, _]}, _query), do: ""
  defp expr({_, _,[_, {{_, _, [{:parent_as, _, _}, _]}, _, _}]}, _query), do: ""
  defp expr({:exists, _, [%Ecto.SubQuery{} = subquery]}, {adapter_meta, _query_meta, query, params, options}) do
    execute(adapter_meta, query, {:all, subquery.query}, params, options)
    |> elem(1)
    |> List.flatten()
    |> Enum.empty?()
    |> Kernel.not()
    |> to_string()
  end
  defp expr({op, _, [left, {o, _, [%Ecto.SubQuery{} = subquery]}]}, {adapter_meta, _query_meta, query, params, options} = q)  when o in ~w[all any]a  do
    values =
     execute(adapter_meta, query, {:all, subquery.query}, params, options)
     |> elem(1)
     |> List.flatten()

     "apply('Elixir.Enum', '#{o}?', [#{expr(values, q)}, fun(Val) -> #{expr(left, q)} #{to_erlang_term(op)} Val end])"
  end
  defp expr({:sum, _, [expr]}, query) do
    "{sum, #{expr(expr, query)}}"
  end
  defp expr({:fragment, _, fragemnt}, query) do
    fragemnt
    |> Enum.reduce([], fn
      {:raw, raw}, acc -> acc ++ [raw]
      {:expr, {expr, _, _}}, acc -> acc ++ [expr(expr, query)]
    end)
    |> to_string()
  end
  defp expr({:not = operator, mdl, [{:in, mdr, [left, %Ecto.SubQuery{} = subquery]}]}, {adapter_meta, _query_meta, query, params, options} = q), do: expr({operator, mdl, [{:in, mdr, [left, elem(execute(adapter_meta, query, {:all, subquery.query}, params, options), 1)]}]}, q)
  defp expr({:not = operator, _, [{:in, _, _} = expr]}, query), do: unroll(expr, query, operator)
  defp expr({:not = operator, [], [{:is_nil, _, [{{:., _, [{:&, _, [index]}, column]}, _, _}]}]}, {adapter_meta, _query_meta, query, _params, _options}) do
    {<<s::binary-size(1), _::binary>>, module, _prefix} = elem(query.sources, index)
    "#{to_element(adapter_meta, column, module.__schema__(:fields), module.__schema__(:primary_key), "#{String.upcase(s)}#{index}")} #{to_erlang_term(operator)} nil"
  end
  defp expr({:not, _, [expr]}, query) do
    "(#{expr(expr, query)}) == false"
  end
  defp expr({operator, _, [{l, _, _} = left, {r, _, _} = right]}, query) when operator in ~w[> < >= <= == === !=]a and l == :datetime_add or r == :datetime_add do
    case operator do
      :>   -> "apply('Elixir.DateTime', compare, [#{expr(left, query)}, #{expr(right, query)}]) == gt"
      :<   -> "apply('Elixir.DateTime', compare, [#{expr(left, query)}, #{expr(right, query)}]) == lt"
      :>=  -> "apply('Elixir.DateTime', compare, [#{expr(left, query)}, #{expr(right, query)}]) == gt or eq"
      :<=  -> "apply('Elixir.DateTime', compare, [#{expr(left, query)}, #{expr(right, query)}]) == lt or eq"
      :==  -> "apply('Elixir.DateTime', compare, [#{expr(left, query)}, #{expr(right, query)}]) == eq"
      :=== -> "apply('Elixir.DateTime', compare, [#{expr(left, query)}, #{expr(right, query)}]) == eq"
      :!=  -> "apply('Elixir.DateTime', compare, [#{expr(left, query)}, #{expr(right, query)}]) /= eq"
    end
  end
  defp expr({operator, _, [left, right]}, query) when operator in ~w[not or and > < >= <= == === != + - * /]a do
    "#{expr(left, query)} #{to_erlang_term(operator)} #{expr(right, query)}"
  end
  defp expr(%Ecto.Query.Tagged{value: expr}, query), do: expr(expr, query)
  defp expr({:json_extract_path, _, [left, right]}, query) do
    "apply('Elixir.EctoQLC.Adapters.QLC', get_in, [#{expr(left, query)}, #{expr(right, query)}])"
  end
  defp expr({:like, _, [left, match]}, query) do
    "apply('Elixir.EctoQLC.Adapters.QLC', 'like', [#{expr(left, query)}, #{expr(match, query)}]) == true"
  end
  defp expr({:ilike, _, [left, match]}, query) do
    "apply('Elixir.EctoQLC.Adapters.QLC', 'ilike', [#{expr(left, query)}, #{expr(match, query)}]) == true"
  end
  defp expr({:datetime_add, _, [left, right, interval]}, query) do
    "apply('Elixir.DateTime', add, [#{expr(left, query)}, #{interval_to_seconds(interval) * expr(right, query)}])"
  end
  defp expr({:date_add, _, [left, right, interval]}, query) do
    "apply('Elixir.Date', add, [#{expr(left, query)}, #{round(interval_to_days(interval) * expr(right, query))}])"
  end
  defp expr({:count, _, [{expr, [], []}, :distinct]}, query), do: expr(expr, query)
  defp expr({:count, _, [{expr, [], []}]}, query), do: expr(expr, query)
  defp expr({:coalesce, _, [left, right]}, query) do
    "apply('Elixir.EctoQLC.Adapters.QLC', coalesce, [#{expr(left, query)}, #{expr(right, query)}])"
  end
  defp expr({:parent_as, _, [key]}, {_adapter_meta, query_meta, _query, _params, _options}), do: query_meta.aliases[key]
  defp expr({:., _, [{:&, _, [idx]}, column]}, {adapter_meta, _query_meta, query, _params, options}) do
    case elem(query.sources, idx) do
      {<<s::binary-size(1), _::binary>> = source, nil, prefix} when adapter_meta.adapter == EctoQLC.Adapters.Mnesia ->
        attributes = :mnesia.table_info(to_table(adapter_meta, source, prefix, options), :attributes)
        to_element(adapter_meta, column, tl(attributes), [hd(attributes)], "#{String.upcase(s)}#{idx}")

      {"schema_migrations", nil, _prefix} ->
        to_element(adapter_meta, column, Ecto.Migration.SchemaMigration.__schema__(:fields), Ecto.Migration.SchemaMigration.__schema__(:primary_key), "S#{idx}")

      {<<s::binary-size(1), _::binary>>, module, _prefix} ->
        to_element(adapter_meta, column, module.__schema__(:fields), module.__schema__(:primary_key), "#{String.upcase(s)}#{idx}")
    end
  end
  defp expr({:., _, [{:parent_as, _, [key]}, column]}, {adapter_meta, query_meta, _query, _params, _options}) do
    idx = query_meta.aliases[key]
    {<<s::binary-size(1), _::binary>>, module, _prefix} = elem(query_meta.sources, query_meta.aliases[key])
    to_element(adapter_meta, column, module.__schema__(:fields), module.__schema__(:primary_key), "#{String.upcase(s)}#{idx}")
  end
  defp expr({:in, metadata, [left, %Ecto.SubQuery{} = subquery]}, {adapter_meta, _query_meta, query, params, options} = q), do: expr({:in, metadata, [left, elem(execute(adapter_meta, query, {:all, subquery.query}, params, options), 1)]}, q)
  defp expr({:in, _, _} = expr, query), do: unroll(expr, query, :==)
  defp expr({:is_nil, _, [expr]}, query), do: "#{expr(expr, query)} == nil"
  defp expr({:^, _, [ix]}, _query), do: "PARAM#{ix + 1}"
  defp expr({expr, [], []}, query), do: expr(expr, query)
  defp expr(expr, _query), do: to_erlang_term(expr)

  defp expr({:filter, _, [expr, filter]}, group, query) do
    expr(expr, Enum.filter(group, &expr(filter, &1, query)), query)
  end
  defp expr({:count, _, []}, group, _query), do: length(group)
  defp expr({:count, _, [expr, :distinct]}, group, query) do
    group
    |> Enum.uniq_by(&expr(expr, &1, query))
    |> Enum.count()
  end
  defp expr({:avg, _, [expr]}, group, query) do
    case Enum.map(group, &expr(expr, &1, query)) do
      [] -> 0
      values -> Enum.sum(values) / length(values)
    end
  end
  defp expr({op, _, [expr]}, group, query) when op in @aggregates do
    case Enum.map(group, &expr(expr, &1, query)) do
      [] -> 0
      values -> apply(Enum, op, [values])
    end
  end
  defp expr({:in, _, [left, right]}, row, query), do: expr(left, row, query) in expr(right, row, query)
  defp expr({:and, _, [left, right]}, row, query), do: expr(left, row, query) and expr(right, row, query)
  defp expr({:or, _, [left, right]}, row, query), do: expr(left, row, query) or expr(right, row, query)
  defp expr({op, _, [left, right]}, row, query) when op in @operators, do: apply(Kernel, op, [expr(left, row, query), expr(right, row, query)])
  defp expr({op, _, _} = expr, [row | _] = group, query) when op not in @aggregates and is_tuple(row), do: expr(expr, List.first(group), query)
  defp expr({:parent_as, _, [key]}, _row, {_adapter_meta, query_meta, _query, _params, _options}), do: query_meta.aliases[key]
  defp expr({:&, _, [idx]}, _row, _query), do: idx
  defp expr({{:., _, [_expr, _column]}, _, _} = expr, row, {adapter_meta, query_meta, %{query: query}, params, options}), do: expr(expr, row, {adapter_meta, query_meta, query, params, options})
  defp expr({{:., _, [expr, column]}, _, _}, row, {_adapter_meta, _query_meta, query, _params, _options} = q) do
    idx = expr(expr, row, q)
    case elem(query.sources, idx) do
      %{query: query} ->
        {_source, module, _prefix} = elem(query.sources, idx)

        base = if idx > 0, do: Enum.reduce(0..idx - 1, 0, fn idx, acc ->
          {_source, module, _prefix} = elem(query.sources, idx)
          length(module.__schema__(:fields)) + acc
        end), else: 0
        elem(row, base + Enum.find_index(module.__schema__(:fields), &(&1 == column)))

      {"schema_migrations", nil, _prefix} -> elem(row, 0)

      {_source, module, _prefix} ->
        base = if idx > 0, do: Enum.reduce(0..idx - 1, 0, fn idx, acc ->
          {_source, module, _prefix} = elem(query.sources, idx)
          length(module.__schema__(:fields)) + acc
        end), else: 0
        elem(row, base + Enum.find_index(module.__schema__(:fields), &(&1 == column)))
    end
  end
  defp expr({:^, _, [idx]}, _row, {_adapter_meta, _query_meta, _query, params, _options}), do: Enum.at(params, idx, idx)
  defp expr({:like, _, [left, right]}, row, query), do: String.match?(expr(left, row, query), Regex.compile!(expr(right, row, query)))
  defp expr({:ilike, _, [left, right]}, row, query), do: String.match?(expr(left, row, query), Regex.compile!(expr(right, row, query), [:caseless]))
  defp expr({:json_extract_path, _, [left, right]}, row, query) do
    Enum.reduce(expr(right, row, query), expr(left, row, query), fn
    _k, nil = data -> data
    k, %_{} = struct -> Map.get(struct, String.to_existing_atom(k))
    k, data when is_integer(k) and is_list(data) -> Enum.at(data, k)
    k, data when is_integer(k) and is_tuple(data) -> elem(data, k)
    k, data -> data[k] || data[String.to_existing_atom(k)]
  end)
  end
  defp expr({operator, md, [%Ecto.Query.Tagged{value: expr}, right, interval]}, row, query) when operator in ~w[datetime_add date_add]a do
    expr({operator, md, [expr, right, interval]}, row, query)
  end
  defp expr({operator, md, [left, %Ecto.Query.Tagged{value: expr}, interval]}, row, query) when operator in ~w[datetime_add date_add]a do
    expr({operator, md, [left, expr, interval]}, row, query)
  end
  defp expr({:datetime_add, _, [left, right, interval]}, row, query) do
    DateTime.add(expr(left, row, query), interval_to_seconds(interval) * expr(right, row, query), :second)
  end
  defp expr({:date_add, _, [left, right, interval]}, row, query) do
    Date.add(expr(left, row, query), round(interval_to_days(interval) * expr(right, row, query)))
  end
  defp expr({op, _, [left, right]}, row, query), do: apply(__MODULE__, op, [expr(left, row, query), expr(right, row, query)])
  defp expr(%Ecto.Query.Tagged{value: expr}, row, query), do: expr(expr, row, query)
  defp expr({_selected_as, expr}, row, query), do: expr(expr, row, query)
  defp expr(%Ecto.SubQuery{} = subquery, _row, {adapter_meta, _query_meta, query, params, options}), do: execute(adapter_meta, query, {:all, subquery.query}, params, options) |> elem(1) |> List.flatten()
  defp expr(expr, _row, _query), do: expr

  @doc false
  def select(rows, :all, {adapter_meta, _query_meta, query, _params, _options} = q) do
    if query.select && has_aggregates(query.select.fields) do
      rows = rows
             |> Enum.group_by(&Enum.map_join(query.group_bys, ":", fn {_column, idx} -> :erlang.element(idx, &1) end))
             |> Map.values()
             |> Enum.map(fn group -> Enum.map(query.select.fields, &expr(&1, group, q)) end)
             |> distinct(query, adapter_meta)
             |> offset(query, 0)
             |> Enum.take(query.limit)

      {length(rows), rows}
    else
      rows = rows
             |> Enum.map(fn row ->
               if query.select do
                 Enum.map(query.select.fields, &expr(&1, row, q))
               else
                 row
               end
              end)
              |> distinct(query, adapter_meta)
              |> offset(query, 0)
              |> Enum.take(query.limit)

      {length(rows), rows}
    end
  end
  def select(rows, :delete_all, {adapter_meta, _query_meta, %Ecto.Query{from: %{source: {source, _module}}} = query, _params, options}) do
    mod = :"#{String.downcase(List.last(Module.split(adapter_meta.adapter)))}"
    prefix = options[:prefix] || query.from.prefix || query.prefix
    table = to_table(adapter_meta, source, prefix, options)
    rows
    |> distinct(query, adapter_meta)
    |> offset(query, 0)
    |> Enum.take(query.limit)
    |> Enum.reduce({0, nil}, fn row, {count, acc} ->
      args = if mod == :mnesia, do: [:erlang.setelement(1, row, table)], else: [table, row]
      if apply(mod, :delete_object, args) do
        {count + 1, acc}
      else
        {count, acc}
      end
    end)
  end
  def select(rows, :update_all, {%{adapter: EctoQLC.Adapters.Mnesia} = adapter_meta, _query_meta, %Ecto.Query{from: %{source: {source, _module}}} = query, _params, options}) do
    {:atomic, value} = :mnesia.transaction(fn ->
      table = to_table(adapter_meta, source, query.from.prefix || query.prefix, options)
      rows
      |> distinct(query, adapter_meta)
      |> offset(query, 0)
      |> Enum.take(query.limit)
      |> Enum.reduce({0, nil}, fn row, {count, acc} ->
        if _row = update(:mnesia, table, row, query) do
          {count + 1, acc}
        else
          {count, acc}
        end
      end)
   end)
   value
  end
  def select(rows, :update_all, {adapter_meta, _query_meta, %Ecto.Query{from: %{source: {source, _module}}} = query, _params, options}) do
    mod = :"#{String.downcase(List.last(Module.split(adapter_meta.adapter)))}"
    table = to_table(adapter_meta, source, query.from.prefix || query.prefix, options)
    rows
    |> distinct(query, adapter_meta)
    |> offset(query, 0)
    |> Enum.take(query.limit)
    |> Enum.reduce({0, nil}, fn row, {count, acc} ->
      if _row = update(mod, table, row, query) do
        {count + 1, acc}
      else
        {count, acc}
      end
    end)
  end

  @doc false
  def get_in(data, keys) do
    Enum.reduce(keys, data, fn
      k, %_{} = struct -> Map.get(struct, String.to_existing_atom(k))
      k, data -> data[k] || data[String.to_existing_atom(k)]
    end)
  end

  defp update(:ets = mod, table, row, query), do: mod.update_element(table, elem(row, 0), Keyword.values(query.updates))
  defp update(:mnesia = mod, _table, row, query), do: mod.write(Enum.reduce(Keyword.values(query.updates), row, fn {idx, value}, row -> :erlang.setelement(idx, row, value) end))
  defp update(:dets = mod, table, row, query), do: mod.insert(table, Enum.reduce(Keyword.values(query.updates), row, fn {idx, value}, row -> :erlang.setelement(idx, row, value) end))

  @doc false
  def delete(mod, adapter_meta, %{source: source, prefix: prefix, schema: schema}, filters, _returning, options) when mod in ~w[dets ets]a do
    table = to_table(adapter_meta, source, prefix, options)
    ms = to_match_spec(adapter_meta, schema, filters)
    {query_time, count} = :timer.tc(mod, :select_delete, [table, ms])
    if 0 < count do
      {:ok, []}
    else
      {:error, :stale}
    end
    |> log(source, "DELETE #{inspect source} #{inspect filters} MATCHSPEC #{inspect ms}", query_time, 0, 0, 0, :delete_all, adapter_meta.telemetry, filters, [], options ++ adapter_meta.opts)
  end
  def delete(:mnesia = mod, adapter_meta, %{source: source, prefix: prefix, schema: schema}, filters, _returning, options) do
    ms = to_match_spec(adapter_meta, schema, filters)
    fun = fn ->
            with table <- to_table(adapter_meta, source, prefix, options),
                 [row] <- mod.select(table, ms),
                   :ok <- mod.delete_object(row) do
              {:ok, []}
            else
              _ -> {:error, :stale}
            end
          end
    {query_time, {:atomic, result}} = :timer.tc(mod, :transaction, [fun])
    log(result, source, "DELETE #{inspect source} #{inspect filters} MATCHSPEC #{inspect ms}", query_time, 0, 0, 0, :delete_all, adapter_meta.telemetry, filters, [], options ++ adapter_meta.opts)
  end

  @doc false
  def insert(:dets = mod, adapter_meta, %{schema: schema, source: source, prefix: prefix}, fields, _on_conflict, returning, options) do
    table = to_table(adapter_meta, source, prefix, options)
    primary_key = schema.__schema__(:primary_key)
    columns = schema.__schema__(:fields)
    key = get_key(primary_key, columns, fields)
    record = to_record({key}, columns, primary_key, fields)
    file = if dir = Application.get_env(mod, :dir), do: Path.join(dir, "#{table}"), else: table
    options = Keyword.take(Keyword.merge([file: '#{file}'],  options), ~w[access auto_save estimated_no_objects file max_no_slots min_no_slots keypos ram_file repair type]a)
    {query_time, result} = with {:ok, ^table} <- mod.open_file(table, options),
                                {query_time, true} <- :timer.tc(mod, :insert_new, [table, record]),
                                :ok <- mod.sync(table) do
                                {query_time, {:ok, Enum.map(returning, &fields[&1])}}
                            else
                              {query_time, false} when is_integer(query_time) -> {query_time, {:invalid, [unique: "primary_key"]}}
                            end
    log(result, source, "INSERT INTO #{inspect source} RETURNING #{inspect returning} #{inspect fields}", query_time, 0, 0, 0, :insert, adapter_meta.telemetry, fields, [], options ++ adapter_meta.opts)
  end
  def insert(:ets = mod, adapter_meta, %{schema: schema, source: source, prefix: prefix}, fields, _on_conflict, returning, options) do
    table = to_table(adapter_meta, source, prefix, options)
    primary_key = schema.__schema__(:primary_key)
    columns = schema.__schema__(:fields)
    key = get_key(primary_key, columns, fields)
    record = to_record({key}, columns, primary_key, fields)

    {query_time, result} = case :timer.tc(mod, :insert_new, [table, record]) do
                              {query_time, true}  -> {query_time, {:ok, Enum.map(returning, &fields[&1])}}
                              {query_time, false} -> {query_time, {:invalid, [unique: "primary_key"]}}
                            end
    log(result, source, "INSERT INTO #{inspect source} RETURNING #{inspect returning} #{inspect fields}", query_time, 0, 0, 0, :insert, adapter_meta.telemetry, fields, [], options ++ adapter_meta.opts)
  end
  def insert(:mnesia = mod, adapter_meta, %{schema: schema, source: source, prefix: prefix}, fields, _on_conflict, returning, options) do
    table = to_table(adapter_meta, source, prefix, options)
    primary_key = schema.__schema__(:primary_key)
    columns = schema.__schema__(:fields)
    key = get_key(primary_key, columns, fields)
    record = to_record({table, key}, columns, primary_key, fields)
    fun  = fn ->
                with [] <- mod.wread({table, key}),
                    :ok <- mod.write(record) do
                  {:ok, []}
                else
                  [_record] -> {:invalid, [unique: "primary_key"]}

                  {:aborted, {:no_exists, ^table}} -> {:invalid, [no_exists: table]}

                  {:aborted, {:bad_type, _}} -> {:invalid, [bad_type: record]}
                end
            end

   {query_time, result} = case :timer.tc(mod, :transaction, [fun]) do
                            {query_time, {:aborted, {:bad_type, ^record}}} -> {query_time, {:invalid, [bad_type: inspect(record)]}}
                            {query_time, {:atomic, :ok}}                   -> {query_time, {:ok, Enum.map(fields, &Enum.map(returning, fn k -> &1[k] end))}}
                            {query_time, {:atomic, result}}                -> {query_time, result}
                          end

    log(result, source, "INSERT INTO #{inspect source} RETURNING #{inspect returning} #{inspect fields}", query_time, 0, 0, 0, :insert, adapter_meta.telemetry, fields, [], options ++ adapter_meta.opts)
  end

  @doc false
  def insert_all(:dets = mod, adapter_meta, %{schema: schema, source: source, prefix: prefix}, _header, rows, _on_conflict, returning, _placeholders, options) do
    table = to_table(adapter_meta, source, prefix, options)
    {query_time, rows} = if is_list(rows), do: {0, rows}, else: :timer.tc(adapter_meta.repo, :all, [rows])
    primary_key = schema.__schema__(:primary_key)
    columns = schema.__schema__(:fields)
    records = for row <- rows do
      key = get_key(primary_key, columns, row)
      to_record({key}, columns, primary_key, row)
    end
    file = if dir = Application.get_env(mod, :dir), do: Path.join(dir, "#{table}"), else: table
    options = Keyword.take(Keyword.merge([file: '#{file}'],  options), ~w[access auto_save estimated_no_objects file max_no_slots min_no_slots keypos ram_file repair type]a)

    {query_time, result}  = with {open_time, {:ok, ^table}} <- :timer.tc(mod, :open_file, [table, options]),
         {insert_time, true} <- :timer.tc(mod, :insert_new, [table, records]),
         {sync_time, :ok} <- :timer.tc(mod, :sync, [table]) do
         result = unless returning == [], do: Enum.map(rows, &Enum.map(returning, fn k -> &1[k] end))
         {query_time + open_time + insert_time + sync_time, {length(records), result}}
    else
      {insert_time, false} when is_integer(insert_time) ->
        {insert_time, {0, nil}}
    end


    log(result, source, "INSERT INTO #{inspect source} RETURNING #{inspect returning} #{inspect rows}", query_time, 0, 0, 0, :insert, adapter_meta.telemetry, rows, [], options ++ adapter_meta.opts)
  end
  def insert_all(:ets = mod, adapter_meta, %{schema: schema, source: source, prefix: prefix}, _header, rows, _on_conflict, returning, _placeholders, options) do
    table = to_table(adapter_meta, source, prefix, options)
    {query_time, rows} = if is_list(rows), do: {0, rows}, else: :timer.tc(adapter_meta.repo, :all, [rows])
    primary_key = schema.__schema__(:primary_key)
    columns = schema.__schema__(:fields)
    records = for row <- rows, do: to_record({get_key(primary_key, columns, row)}, columns, primary_key, row)
    result = unless returning == [], do: Enum.map(rows, &Enum.map(returning, fn k -> &1[k] end))
    {insert_time, value} = :timer.tc(mod, :insert_new, [table, records])
    query_time = query_time + insert_time
    if value do
      {length(records), result}
    else
      {0, result}
    end
    |> log(source, "INSERT INTO #{inspect source} RETURNING #{inspect returning} #{inspect rows}", query_time, 0, 0, 0, :insert, adapter_meta.telemetry, rows, [], options ++ adapter_meta.opts)
  end
  def insert_all(:mnesia = mod, adapter_meta, %{schema: schema, source: source, prefix: prefix}, _header, rows, _on_conflict, returning, _placeholders, options) do
    table = to_table(adapter_meta, source, prefix, options)
    {query_time, rows} = if is_list(rows), do: {0, rows}, else: :timer.tc(adapter_meta.repo, :all, [rows])
    primary_key = schema.__schema__(:primary_key)
    columns = schema.__schema__(:fields)
    {:atomic, result} = mod.transaction(fn ->
      Enum.reduce(rows, {0, []}, fn row, {count, acc} ->
        key = get_key(primary_key, columns, row)
        record = to_record({table, key}, columns, primary_key, row)
        {insert_time, value} = :timer.tc(mod, :write, [record])
        query_time = query_time + insert_time
        if value == :ok do
          acc = if returning != [], do: acc ++ [Enum.map(returning, fn k -> row[k] end)]
          log({count + 1, acc}, source, "INSERT INTO #{inspect source} RETURNING #{inspect returning} #{inspect row}", query_time, 0, 0, 0, :insert, adapter_meta.telemetry, rows, [], options ++ adapter_meta.opts)
        else
          {count, acc}
        end
      end)
    end)
    result
  end

  @doc false
  def update(:dets = mod, adapter_meta, %{schema: schema, source: source, prefix: prefix}, fields, params, returning, options) do
    table = to_table(adapter_meta, source, prefix, options)
    key = to_key(params)
    file = if dir = Application.get_env(mod, :dir), do: Path.join(dir, "#{table}"), else: table
    options = Keyword.take(Keyword.merge([file: '#{file}'],  options), ~w[access auto_save estimated_no_objects file max_no_slots min_no_slots keypos ram_file repair type]a)
    {query_time, result} = with {:ok, ^table} <- mod.open_file(table, options),
                                        [row] <- mod.lookup(table, key),
                            {query_time, :ok} <- :timer.tc(mod, :insert, [table, update_row(adapter_meta, schema, fields, row)]),
                                          :ok <- mod.sync(table),
                           {decode_time, row} <- :timer.tc(Enum, :map, [returning, &fields[&1]]) do
                             {query_time + decode_time, {:ok, row}}
                           else
                             {query_time, _error} when is_integer(query_time) ->  {query_time, {:invalid, [unique: "primary_key"]}}
                           end
    log(result, source, "UPDATE INTO #{inspect source} RETURNING #{inspect returning} #{inspect fields}", query_time, 0, 0, 0, :update_all, adapter_meta.telemetry, params, [], options ++ adapter_meta.opts)
  end
  def update(:ets = mod, adapter_meta, %{schema: schema, source: source, prefix: prefix}, fields, params, returning, options) do
    key = to_key(params)
    {records, _count} = Enum.reduce(schema.__schema__(:fields) -- schema.__schema__(:primary_key), {[], 1}, fn k, {acc, count} -> if Keyword.has_key?(fields, k), do: {[{count, fields[k]} | acc], count + 1}, else: {acc, count + 1}  end)
    table = to_table(adapter_meta, source, prefix, options)
    {query_time, result} = with {query_time, true} <- :timer.tc(mod, :update_element, [table, key, records]),
                                {decode_time, row} <- :timer.tc(Enum, :map, [returning, &fields[&1]]) do
                             {query_time + decode_time, {:ok, row}}
                           else
                             {query_time, _} -> {query_time, {:invalid, [unique: "primary_key"]}}
                           end
    log(result, source, "UPDATE INTO #{inspect source} RETURNING #{inspect returning} #{inspect fields}", query_time, 0, 0, 0, :update_all, adapter_meta.telemetry, params, [], options ++ adapter_meta.opts)
  end
  def update(:mnesia = mod, adapter_meta, %{schema: schema, source: source, prefix: prefix}, fields, params, returning, options) do
   table = to_table(adapter_meta, source, prefix, options)
   key = to_key(params)
   fun = fn ->
            with [row] <- mod.wread({table, key}),
            :ok <- mod.write(update_row(adapter_meta, schema, fields, row)) do
              {:ok, []}
            else
            {:error, _resaon} ->
              {:invalid, [unique: "primary_key"]}
            end
          end

   {query_time, result} = case :timer.tc(mod, :transaction, [fun]) do
                            {query_time, {:atomic, :ok}}     -> {query_time, {:ok, Enum.map(fields, &Enum.map(returning, fn k -> &1[k] end))}}
                            {query_time, {:atomic, result}}  -> {query_time, result}
                          end
   log(result, source, "UPDATE INTO #{inspect source} RETURNING #{inspect returning} #{inspect fields}", query_time, 0, 0, 0, :update_all, adapter_meta.telemetry, params, [], options ++ adapter_meta.opts)
 end

  defp distinct(rows, %Ecto.Query{distinct: nil}, _adapter_meta), do: rows
  defp distinct(rows, %Ecto.Query{distinct: true}, _adapter_meta), do: Enum.uniq(rows)
  defp distinct(rows, %Ecto.Query{distinct: %Ecto.Query.QueryExpr{expr: true}}, _adapter_meta), do: Enum.uniq(rows)
  defp distinct(rows, %Ecto.Query{distinct: %Ecto.Query.QueryExpr{expr: expr}} = query, adapter_meta) do
    Enum.reduce(expr, rows, fn
      {sorter, {{:., _, [{:&, _, [index]}, column]}, _, _}}, rows ->
        {_source, module, _prefix} = elem(query.sources, index)
        idx = get_index(adapter_meta, column, module.__schema__(:fields), module.__schema__(:primary_key))
        rows
        |> Enum.sort_by(fn
          [v] -> v
          row -> Enum.at(row, idx)
        end, sorter)
        |> Enum.uniq_by(fn
          [v] -> v
          row -> Enum.at(row, idx)
        end)

      {sorter, {:json_extract_path, _, [{{:., _, [{:&, _, [index]}, column]}, _, _}, keys]}}, rows ->
          {_source, module, _prefix} = elem(query.sources, index)
          idx = get_index(adapter_meta, column, module.__schema__(:fields), module.__schema__(:primary_key))

          rows
          |> Enum.sort_by(fn
            [v] -> v
            row ->
              Enum.reduce(keys, Enum.at(row, idx), fn
                k, %_{} = struct -> Map.get(struct, String.to_existing_atom(k))
                k, data -> data[k] || data[String.to_existing_atom(k)]
              end)
          end, sorter)
          |> Enum.uniq_by(fn
            [v] -> v
            row ->
              Enum.reduce(keys, Enum.at(row, idx), fn
                k, %_{} = struct -> Map.get(struct, String.to_existing_atom(k))
                k, data -> data[k] || data[String.to_existing_atom(k)]
              end)
          end)
      end)
  end

  defp offset(rows, %{offset: offset}, offset), do: rows
  defp offset([_ | rows], query, offset), do: offset(rows, query, offset + 1)

  ## TBD There is a cavet here and we should properly use ex_cldr to do proper calculation
  ## E.g months in second is the same as 30 days in seconds, but some months don't have 30 days.
  defp interval_to_seconds("year"), do: 31557600
  defp interval_to_seconds("month"), do: 2629800
  defp interval_to_seconds("week"), do: 604800
  defp interval_to_seconds("day"), do: 86400
  defp interval_to_seconds("hour"), do: 3600
  defp interval_to_seconds("minute"), do: 60
  defp interval_to_seconds("seconds"), do: 1
  defp interval_to_seconds("millisecond"), do: 0.001
  defp interval_to_seconds("microsecond"), do: 0.000001

  defp interval_to_days("year"), do: 365.25
  defp interval_to_days("month"), do: 30.4375
  defp interval_to_days("week"), do: 7
  defp interval_to_days("day"), do: 1
  defp interval_to_days("hour"), do: 0.04166667
  defp interval_to_days("minute"), do: 6.944444e-4
  defp interval_to_days("seconds"), do: 1.157407e-5
  defp interval_to_days("millisecond"), do: 1.157407e-8
  defp interval_to_days("microsecond"), do: 1.157407e-11

  defp unroll({:in, _, [{{:., [], [{:&, [], [index]}, column]}, _, _}, values]}, {adapter_meta, _query_meta, query, _params, _options} = q, operator) do
    [v | values] = unbind(values, q)
    {<<s::binary-size(1), _::binary>>, module, _prefix} = elem(query.sources, index)
    el = to_element(adapter_meta, column, module.__schema__(:fields), module.__schema__(:primary_key), "#{String.upcase(s)}#{index}")
    values
    |> Enum.reduce(["#{el} #{to_erlang_term(operator)} #{v}"], fn v, acc -> acc ++ [" orelse #{el} #{to_erlang_term(operator)} #{v}"] end)
    |> to_string()
  end

  defp unbind({:^, _, [index]}, {_adapter_meta, _query_meta, _query, params, _options}), do: to_erlang_term(Enum.at(params, index))
  defp unbind({:^, _, values}, {_adapter_meta, _query_meta, _query, params, _options}), do: Enum.map(values, &to_erlang_term(Enum.at(params, &1, &1)))
  defp unbind([_ | _] = values, query), do: Enum.map(values, &unbind(&1, query))
  defp unbind(value, _meta), do: to_erlang_term(value)

  defp to_erlang_term(<<value::binary>>), do: '<<"#{value}">>'
  defp to_erlang_term(:or), do: :orelse
  defp to_erlang_term(:and), do: :andalso
  defp to_erlang_term(:<=), do: :"=<"
  defp to_erlang_term(:===), do: :"=:="
  defp to_erlang_term(op) when op in ~w[!= not]a, do: :"/="
  defp to_erlang_term(op) when op in ~w[> < >= == + - * /]a, do: "#{op}"
  defp to_erlang_term(value) when is_list(value), do: "[#{Enum.map_join(value, ", ", &to_erlang_term(&1))}]"
  defp to_erlang_term(value) when is_tuple(value), do: "#{value |> Tuple.to_list() |> Enum.map(&to_erlang_term/1) |> List.to_tuple() |> inspect}"
  defp to_erlang_term(value) when is_atom(value), do: "'#{value}'"
  defp to_erlang_term(value) when is_number(value), do: value
  defp to_erlang_term(value) when is_map(value), do: "#" <> "{" <> Enum.map_join(value, ", ", fn {k, v} -> "#{to_erlang_term(k)} => #{to_erlang_term(v)}" end) <> "}"
  defp to_erlang_term(value), do: value


  defp to_record(tuple, [_ | columns], [], fields), do: Enum.reduce(columns, tuple, &Tuple.insert_at(&2, tuple_size(&2), fields[&1]))
  defp to_record(tuple, columns, primary_key, fields), do: Enum.reduce(columns -- primary_key, tuple, &Tuple.insert_at(&2, tuple_size(&2), fields[&1]))

  defp to_element(adapter_meta, column, columns, [_primary_key] = primary_keys, var) do
    "element(#{get_index(adapter_meta, column, columns, primary_keys)}, #{var})"
  end
  defp to_element(adapter_meta, column, columns, primary_keys, var) do
   idx = get_index(adapter_meta, column, columns, primary_keys)
   if column in primary_keys do
      "element(#{idx}, element(#{if adapter_meta.adapter == EctoQLC.Adapters.Mnesia, do: 2, else: 1}, #{var}))"
   else
     "element(#{idx}, #{var})"
    end
  end

  defp get_index(adapter_meta, column, columns, [_primary_key]) do
    Enum.find_index(columns, &(&1 == column))
    |> Kernel.||(0)
    |> Kernel.+(if(adapter_meta.adapter == EctoQLC.Adapters.Mnesia, do: 2, else: 1))
  end
  defp get_index(adapter_meta, column, columns, primary_keys) do
    if column in primary_keys do
      Enum.find_index(primary_keys, &(&1 == column)) + 1
    else
      Enum.find_index(columns -- primary_keys, &(&1 == column)) + length(primary_keys) + if(adapter_meta.adapter == EctoQLC.Adapters.Mnesia, do: 1, else: 1)
    end
  end

  defp to_key(params) do
    case Keyword.values(params) do
      [k] -> k
      values -> List.to_tuple(values)
    end
  end

  defp to_table(adapter_meta, source, prefix, options) do
    Module.concat([adapter_meta.adapter, options[:prefix] || prefix, source])
  end

  defp log(:ok, source, query, query_time, decode_time, queue_time, idle_time, operator, telemetry, params, columns, opts) do
    log({0, []}, source, query, query_time, decode_time, queue_time, idle_time, operator, telemetry, params, columns, opts)
  end
  defp log({num_rows, rows} = result, source, query, query_time, decode_time, queue_time, idle_time, operator, {repo, log, event_name} = _telemetry, params, columns, opts) do
    columns = if is_struct(columns) and columns.select, do: columns.select.fields, else: []
    query = String.Chars.to_string(query)
    stacktrace = Keyword.get(opts, :stacktrace)
    if event_name = Keyword.get(opts, :telemetry_event, event_name) do
      :telemetry.execute(event_name,
                        %{query_time: query_time, decode_time: decode_time, queue_time: queue_time, idle_time: idle_time, total_time: query_time + decode_time + queue_time + idle_time},
                        %{type: :ecto_qlc_query, repo: repo, result: {:ok, %{command: to_command(operator), rows: rows, num_rows: num_rows, columns: columns}}, params: params, query: query, source: source, stacktrace: stacktrace, options: Keyword.get(opts, :telemetry_options, [])})
    end
    fun = fn -> log_iodata(query_time, decode_time, queue_time, idle_time, repo, source, query, opts[:cast_params] || params, result, stacktrace) end
    case Keyword.get(opts, :log, log) do
      true  ->
        Logger.log(:debug, fun, ansi_color: operator_to_color(operator))
        result
      false ->
        :ok
        result
      level ->
        Logger.log(level, fun, ansi_color: operator_to_color(operator))
        result
    end
  end

  defp to_command(:all), do: :select
  defp to_command(command), do: command

  defp log_iodata(query_time, decode_time, queue_time, idle_time, repo, source, query, params, result, stacktrace) do
    result =  if is_tuple(result) and is_atom(elem(result, 0)), do: String.upcase("#{elem(result, 0)}"), else: "OK"
    stacktrace = case stacktrace do
      [_ | _] = stacktrace ->
        {module, function, arity, info} = last_non_ecto(Enum.reverse(stacktrace), repo, nil)
        [?\n, IO.ANSI.light_black(), "↳ ", Exception.format_mfa(module, function, arity), log_stacktrace_info(info), IO.ANSI.reset()]
        _ -> []
    end
    ['QUERY ',
     result,
     " source=#{inspect(source)}",
     format_time("db", query_time),
     format_time("decode", decode_time),
     format_time("queue", queue_time),
     format_time("idle_time", idle_time),
     ?\n,
     query,
     ?\s,
     inspect(params, charlists: false),
     List.wrap(stacktrace)]
  end

  defp format_time(label, time) when time > 999, do: [?\s, label, ?=, :io_lib_format.fwrite_g(time / 1000), ?m, ?s]
  defp format_time(label, time) when time <= 999, do: [?\s, label, ?=, "#{time}", ?μ, ?s]

  defp log_stacktrace_info([file: file, line: line] ++ _rest), do: [", at: ", file, ?:, Integer.to_string(line)]
  defp log_stacktrace_info(_), do: []

  defp last_non_ecto([{mod, _, _, _} | _stacktrace], repo, last) when mod == repo or mod in [Ecto.Repo.Queryable, Ecto.Repo.Schema, Ecto.Repo.Transaction], do: last
  defp last_non_ecto([last | stacktrace], repo, _last), do: last_non_ecto(stacktrace, repo, last)
  defp last_non_ecto([], _repo, last), do: last

  defp operator_to_color(:all), do: :cyan
  defp operator_to_color(:update_all), do: :yellow
  defp operator_to_color(:delete_all), do: :red
  defp operator_to_color(:insert), do: :green
  defp operator_to_color(_op), do: nil
end