if Code.ensure_loaded?(Ecto.Adapters.SQL) do defmodule Ecto.Adapters.QuackDB.AppendInsert do @moduledoc false def run(adapter_meta, schema_meta, header, rows, on_conflict, returning, placeholders, opts) do with :ok <- assert_supported!(schema_meta, rows, on_conflict, returning, placeholders, opts), conn <- ecto_connection(adapter_meta), append_header <- append_header(schema_meta, header), options <- append_options(schema_meta, append_header, opts), {:ok, %QuackDB.Result{} = result} <- insert_all(conn, schema_meta, append_header, rows, on_conflict, returning, options) do {result.num_rows, result.rows} else {:error, %QuackDB.Error{} = error} -> raise error end end defp insert_all(conn, schema_meta, header, rows, on_conflict, returning, options) do if insert_select?(schema_meta, header, on_conflict, returning) do insert_select(conn, schema_meta, header, rows, on_conflict, returning, options) else direct_insert_all(conn, schema_meta, header, rows, options) end end defp direct_insert_all(conn, schema_meta, header, rows, options) do case Keyword.get(options, :append_shape, :columns) do :columns -> columns = append_columns(header, rows, options) QuackDB.insert_columns(conn, schema_meta.source, columns, options) :rows -> rows = append_rows(header, rows, options) QuackDB.insert_rows(conn, schema_meta.source, rows, options) end end defp insert_select?(_schema_meta, _header, _on_conflict, returning) when returning != [], do: true defp insert_select?(_schema_meta, _header, {:nothing, _params, _targets}, _returning), do: true defp insert_select?(%{schema: nil}, _header, _on_conflict, _returning), do: false defp insert_select?(schema_meta, header, _on_conflict, _returning) do MapSet.new(header) != MapSet.new(schema_source_order(schema_meta)) end defp insert_select(conn, schema_meta, header, rows, on_conflict, returning, options) do case DBConnection.status(conn, options) do :idle -> DBConnection.transaction( conn, fn tx -> case do_insert_select( tx, schema_meta, header, rows, on_conflict, returning, options ) do {:ok, result} -> result {:error, error} -> DBConnection.rollback(tx, error) end end, options ) _status -> do_insert_select(conn, schema_meta, header, rows, on_conflict, returning, options) end end defp do_insert_select(conn, schema_meta, header, rows, on_conflict, returning, options) do temp_columns = append_columns!(schema_meta, header, options) temp_table = temp_table_name(temp_columns) create_statement = create_temp_table(temp_table, temp_columns) clear_statement = clear_temp_table(temp_table) insert_statement = insert_from_temp_statement(schema_meta, temp_columns, temp_table, on_conflict, returning) try do with {:ok, _result} <- QuackDB.query(conn, create_statement, [], options), {:ok, _result} <- QuackDB.query(conn, clear_statement, [], options), {:ok, _result} <- append_temp_data(conn, temp_table, header, rows, options), {:ok, %QuackDB.Result{} = result} <- QuackDB.query(conn, insert_statement, [], options) do {:ok, result} end after _ = QuackDB.query(conn, clear_statement, [], options) end end defp temp_table_name(columns) do hash = columns |> :erlang.phash2(4_294_967_296) |> Integer.to_string(36) "quackdb_append_#{hash}" end defp append_columns!(_schema_meta, header, options) do types = Keyword.fetch!(options, :columns) Enum.map(header, fn source -> %{ source: source, type: column_type!(types, source) } end) end defp append_header(%{schema: nil}, header), do: header defp append_header(schema_meta, header) do schema_sources = schema_source_order(schema_meta) if MapSet.new(header) == MapSet.new(schema_sources), do: schema_sources, else: header end defp schema_source_order(%{schema: schema}) do Enum.map(schema.__schema__(:fields), &schema.__schema__(:field_source, &1)) end defp create_temp_table(temp_table, columns) do ddl_columns = Enum.map(columns, fn column -> {column.source, column.type} end) QuackDB.DDL.create_table(temp_table, ddl_columns, temporary: true, if_not_exists: true) end defp clear_temp_table(temp_table) do ["DELETE FROM ", QuackDB.Type.quote_identifier(temp_table)] end defp column_type!(columns, column) do Enum.find_value(columns, fn {^column, type} -> {:ok, type} {name, type} when is_atom(name) and is_binary(column) -> if Atom.to_string(name) == column, do: {:ok, type} {name, type} when is_binary(name) and is_atom(column) -> if name == Atom.to_string(column), do: {:ok, type} _entry -> nil end) |> case do {:ok, type} -> type nil -> raise KeyError, key: column, term: columns end end defp append_temp_data(conn, temp_table, header, rows, options) do case Keyword.get(options, :append_shape, :columns) do :columns -> columns = append_columns(header, rows, options) QuackDB.insert_columns(conn, temp_table, columns, options) :rows -> rows = append_rows(header, rows, options) QuackDB.insert_rows(conn, temp_table, rows, options) end end defp insert_from_temp_statement(schema_meta, columns, temp_table, on_conflict, returning) do header = Enum.map(columns, & &1.source) QuackDB.DML.insert_into_select( {schema_meta.prefix, schema_meta.source}, header, temp_table, header, on_conflict: dml_on_conflict(on_conflict), returning: returning ) end defp dml_on_conflict({:raise, _params, []}), do: :raise defp dml_on_conflict({:nothing, _params, []}), do: :nothing defp dml_on_conflict({:nothing, _params, targets}), do: {:nothing, targets} defp assert_supported!( _schema_meta, {%Ecto.Query{}, _params}, _on_conflict, _returning, _placeholders, _opts ) do unsupported!( :schema_inserts, "insert_method: :append does not support insert_all from queries" ) end defp assert_supported!( _schema_meta, %Ecto.Query{}, _on_conflict, _returning, _placeholders, _opts ) do unsupported!( :schema_inserts, "insert_method: :append does not support insert_all from queries" ) end defp assert_supported!(schema_meta, _rows, {:raise, _params, []}, returning, [], opts) when is_list(returning) do assert_insert_select_columns!(schema_meta, returning != [], opts, "returning") end defp assert_supported!(schema_meta, _rows, {:nothing, _params, _targets}, returning, [], opts) when is_list(returning) do assert_insert_select_columns!(schema_meta, true, opts, "on_conflict: :nothing") end defp assert_supported!(_schema_meta, _rows, _on_conflict, _returning, placeholders, _opts) when placeholders != [] do unsupported!( :schema_inserts, "insert_method: :append does not support placeholders" ) end defp assert_supported!(_schema_meta, _rows, _on_conflict, _returning, _placeholders, _opts) do unsupported!( :schema_inserts, "insert_method: :append only supports plain insert_all or on_conflict: :nothing" ) end defp assert_insert_select_columns!(_schema_meta, false, _opts, _feature), do: :ok defp assert_insert_select_columns!(schema_meta, true, opts, feature) do if schema_meta.schema != nil or Keyword.has_key?(opts, :columns) do :ok else unsupported!( :schema_inserts, "insert_method: :append with #{feature} requires a schema or explicit append columns" ) end end defp ecto_connection(%{pid: pool} = adapter_meta) do case Process.get({Ecto.Adapters.SQL, pool}) do :undefined -> ecto_pool(adapter_meta) nil -> ecto_pool(adapter_meta) conn -> conn end end defp ecto_pool(%{partition_supervisor: {name, _}}), do: {:via, PartitionSupervisor, {name, self()}} defp ecto_pool(%{pid: pool}), do: pool defp append_rows(header, rows, options) do column_types = Enum.map(header, &column_type(Keyword.get(options, :columns, []), &1)) Enum.map(rows, fn row -> case ordered_row_entries(row, header, column_types, []) do {:ok, entries} -> entries :error -> fetched_row_entries(row, header, column_types) end end) end defp ordered_row_entries([], [], [], entries), do: {:ok, Enum.reverse(entries)} defp ordered_row_entries( [{field, value} | row], [field | header], [type | column_types], entries ) do entry = {field, normalize_append_value(value, type)} ordered_row_entries(row, header, column_types, [entry | entries]) end defp ordered_row_entries(_row, _header, _column_types, _entries), do: :error defp fetched_row_entries(row, header, column_types) do header |> Enum.zip(column_types) |> Enum.map(fn {field, type} -> {field, row |> Keyword.fetch!(field) |> normalize_append_value(type)} end) end defp append_columns(header, rows, options) do column_types = Enum.map(header, &column_type(Keyword.get(options, :columns, []), &1)) accumulators = Enum.map(header, fn _field -> [] end) rows |> Enum.reduce(accumulators, fn row, accumulators -> row |> append_row_values(header, column_types) |> prepend_column_values(accumulators) end) |> then(fn columns -> header |> Enum.zip(columns) |> Enum.map(fn {field, values} -> {field, Enum.reverse(values)} end) end) end defp append_row_values(row, header, column_types) do case ordered_row_values(row, header, column_types, []) do {:ok, values} -> values :error -> fetched_row_values(row, header, column_types) end end defp ordered_row_values([], [], [], values), do: {:ok, Enum.reverse(values)} defp ordered_row_values( [{field, value} | row], [field | header], [type | column_types], values ) do ordered_row_values(row, header, column_types, [normalize_append_value(value, type) | values]) end defp ordered_row_values(_row, _header, _column_types, _values), do: :error defp fetched_row_values(row, header, column_types) do header |> Enum.zip(column_types) |> Enum.map(fn {field, type} -> row |> Keyword.fetch!(field) |> normalize_append_value(type) end) end defp prepend_column_values(values, accumulators) do values |> Enum.zip(accumulators) |> Enum.map(fn {value, accumulator} -> [value | accumulator] end) end defp normalize_append_value(nil, _type), do: nil defp normalize_append_value(value, :varchar) when is_map(value) and not is_struct(value) do JSON.encode!(value) end defp normalize_append_value(value, _type), do: value defp column_type(columns, column) do case column_type!(columns, column) do {:json, _type} -> :varchar type -> type end rescue KeyError -> nil end defp append_options(schema_meta, header, opts) do opts = opts |> base_options() |> maybe_put_schema(schema_meta) case Keyword.fetch(opts, :columns) do {:ok, _columns} -> opts :error -> maybe_put_schema_columns(opts, schema_meta, header) end end defp maybe_put_schema(opts, %{prefix: nil}), do: opts defp maybe_put_schema(opts, %{prefix: prefix}), do: Keyword.put(opts, :schema, prefix) defp maybe_put_schema_columns(opts, %{schema: nil}, _header), do: opts defp maybe_put_schema_columns(opts, %{schema: schema}, header) do source_types = Map.new(schema.__schema__(:fields), fn field -> {schema.__schema__(:field_source, field), QuackDB.Ecto.Type.column_type!(schema.__schema__(:type, field), :append)} end) columns = Enum.map(header, fn source -> {source, Map.fetch!(source_types, source)} end) Keyword.put(opts, :columns, columns) end defp base_options(opts) do opts |> Keyword.take([:timeout, :columns, :append_shape]) |> validate_append_shape!() |> maybe_put_batch_size(opts) end defp validate_append_shape!(options) do case Keyword.get(options, :append_shape, :columns) do shape when shape in [:columns, :rows] -> options other -> unsupported!( :schema_inserts, "unsupported append_shape for QuackDB append insert: #{inspect(other)}" ) end end defp maybe_put_batch_size(options, opts) do case Keyword.fetch(opts, :chunk_every) do {:ok, chunk_every} -> Keyword.put(options, :batch_size, chunk_every) :error -> options end end defp unsupported!(feature, message) do raise QuackDB.Error.new(:ecto_feature_not_supported, message, source: :client, metadata: %{feature: feature} ) end end end