Allow insert_all to be called with a query instead of rows (#308)

Author: narrowtux

lib/ecto/adapters/myxql/connection.ex

@@ -142,7 +142,7 @@ if Code.ensure_loaded?(MyXQL) do
     @impl true
     def insert(prefix, table, header, rows, on_conflict, [], []) do
       fields = quote_names(header)
-      ["INSERT INTO ", quote_table(prefix, table), " (", fields, ") VALUES ",
+      ["INSERT INTO ", quote_table(prefix, table), " (", fields, ") ",
        insert_all(rows) | on_conflict(on_conflict, header)]
     end
     def insert(_prefix, _table, _header, _rows, _on_conflict, _returning, []) do
@@ -151,7 +151,7 @@ if Code.ensure_loaded?(MyXQL) do
     def insert(_prefix, _table, _header, _rows, _on_conflict, _returning, _placeholders) do
       error!(nil, ":placeholders is not supported by MySQL")
     end
-    
+
     defp on_conflict({_, _, [_ | _]}, _header) do
       error!(nil, "The :conflict_target option is not supported in insert/insert_all by MySQL")
     end
@@ -176,10 +176,14 @@ if Code.ensure_loaded?(MyXQL) do
       error!(nil, "Using a query with :where in combination with the :on_conflict option is not supported by MySQL")
     end
 
-    defp insert_all(rows) do
-      intersperse_map(rows, ?,, fn row ->
+    defp insert_all(rows) when is_list(rows) do
+      ["VALUES ", intersperse_map(rows, ?,, fn row ->
         [?(, intersperse_map(row, ?,, &insert_all_value/1), ?)]
-      end)
+      end)]
+    end
+
+    defp insert_all(%Ecto.Query{} = query) do
+      [?(, all(query), ?)]
     end
 
     defp insert_all_value(nil), do: "DEFAULT"

lib/ecto/adapters/postgres/connection.ex

@@ -158,7 +158,7 @@ if Code.ensure_loaded?(Postgrex) do
         if header == [] do
           [" VALUES " | intersperse_map(rows, ?,, fn _ -> "(DEFAULT)" end)]
         else
-          [?\s, ?(, quote_names(header), ") VALUES " | insert_all(rows, counter_offset)]
+          [" (", quote_names(header), ") " | insert_all(rows, counter_offset)]
         end
 
       ["INSERT INTO ", quote_table(prefix, table), insert_as(on_conflict),
@@ -199,12 +199,16 @@ if Code.ensure_loaded?(Postgrex) do
        end)]
     end
 
+    defp insert_all(query = %Ecto.Query{}, _counter) do
+      [?(, all(query), ?)]
+    end
+
     defp insert_all(rows, counter) do
-      intersperse_reduce(rows, ?,, counter, fn row, counter ->
+      ["VALUES ", intersperse_reduce(rows, ?,, counter, fn row, counter ->
         {row, counter} = insert_each(row, counter)
         {[?(, row, ?)], counter}
       end)
-      |> elem(0)
+      |> elem(0)]
     end
 
     defp insert_each(values, counter) do
@@ -217,7 +221,7 @@ if Code.ensure_loaded?(Postgrex) do
 
         {:placeholder, placeholder_index}, counter ->
           {[?$ | placeholder_index], counter}
-          
+
         _, counter ->
           {[?$ | Integer.to_string(counter)], counter + 1}
       end)

lib/ecto/adapters/sql.ex

@@ -642,13 +642,16 @@ defmodule Ecto.Adapters.SQL do
   def insert_all(adapter_meta, schema_meta, conn, header, rows, on_conflict, returning, placeholders, opts) do
     %{source: source, prefix: prefix} = schema_meta
     {_, conflict_params, _} = on_conflict
-    {rows, params} = unzip_inserts(header, rows)
+    {rows, params} = case rows do
+      {%Ecto.Query{} = query, params} -> {query, Enum.reverse(params)}
+      rows -> unzip_inserts(header, rows)
+    end
 
     sql = conn.insert(prefix, source, header, rows, on_conflict, returning, placeholders)
 
     opts = [{:cache_statement, "ecto_insert_all_#{source}"} | opts]
 
-    all_params = placeholders ++ Enum.reverse(params,  conflict_params)
+    all_params = placeholders ++ Enum.reverse(params, conflict_params)
     %{num_rows: num, rows: rows} =
       query!(adapter_meta, sql, all_params, opts)
 

lib/ecto/adapters/tds/connection.ex

@@ -212,8 +212,8 @@ if Code.ensure_loaded?(Tds) do
             ?(,
             quote_names(header),
             ?),
-            returning,
-            " VALUES " | insert_all(rows, counter_offset)
+            returning |
+            insert_all(rows, counter_offset)
           ]
         end
 
@@ -228,12 +228,18 @@ if Code.ensure_loaded?(Tds) do
       error!(nil, "Tds adapter supports only on_conflict: :raise")
     end
 
+    defp insert_all(%Ecto.Query{} = query, _counter) do
+      [?\s, all(query)]
+    end
     defp insert_all(rows, counter) do
-      intersperse_reduce(rows, ",", counter, fn row, counter ->
-        {row, counter} = insert_each(row, counter)
-        {[?(, row, ?)], counter}
-      end)
-      |> elem(0)
+      sql =
+        intersperse_reduce(rows, ",", counter, fn row, counter ->
+          {row, counter} = insert_each(row, counter)
+          {[?(, row, ?)], counter}
+        end)
+        |> elem(0)
+
+      [" VALUES " | sql]
     end
 
     defp insert_each(values, counter) do

test/ecto/adapters/myxql_test.exs

@@ -974,6 +974,13 @@ defmodule Ecto.Adapters.MyXQLTest do
     assert query == ~s{INSERT INTO `schema` (`x`,`y`,`z`) VALUES (?,(SELECT s0.`id` FROM `schema` AS s0),?),(DEFAULT,DEFAULT,(SELECT s0.`id` FROM `schema` AS s0))}
   end
 
+  test "insert with query as rows" do
+    query = from(s in "schema", select: %{ foo: fragment("3"), bar: s.bar }) |> plan(:all)
+    query = insert(nil, "schema", [:foo, :bar], query, {:raise, [], []}, [])
+
+    assert query == ~s{INSERT INTO `schema` (`foo`,`bar`) (SELECT 3, s0.`bar` FROM `schema` AS s0)}
+  end
+
   test "update" do
     query = update(nil, "schema", [:id], [x: 1, y: 2], [])
     assert query == ~s{UPDATE `schema` SET `id` = ? WHERE `x` = ? AND `y` = ?}

test/ecto/adapters/postgres_test.exs

@@ -1147,6 +1147,13 @@ defmodule Ecto.Adapters.PostgresTest do
     assert query == ~s{INSERT INTO "schema" ("x","y","z") VALUES ($1,(SELECT s0."id" FROM "schema" AS s0),$5),(DEFAULT,(SELECT s0."id" FROM "schema" AS s0),$8) RETURNING "id"}
   end
 
+  test "insert with query as rows" do
+    query = from(s in "schema", select: %{ foo: fragment("3"), bar: s.bar }) |> plan(:all)
+    query = insert(nil, "schema", [:foo, :bar], query, {:raise, [], []}, [:foo])
+
+    assert query == ~s{INSERT INTO "schema" ("foo","bar") (SELECT 3, s0."bar" FROM "schema" AS s0) RETURNING "foo"}
+  end
+
   test "update" do
     query = update(nil, "schema", [:x, :y], [id: 1], [])
     assert query == ~s{UPDATE "schema" SET "x" = $1, "y" = $2 WHERE "id" = $3}

test/ecto/adapters/tds_test.exs

@@ -1070,6 +1070,13 @@ defmodule Ecto.Adapters.TdsTest do
              ~s{INSERT INTO [schema] ([x],[y],[z]) VALUES (@1, (SELECT s0.[id] FROM [schema] AS s0), @4),(DEFAULT, DEFAULT, (SELECT s0.[id] FROM [schema] AS s0))}
   end
 
+  test "insert with query as rows" do
+    query = from(s in "schema", select: %{ foo: fragment("3"), bar: s.bar }) |> plan(:all)
+    query = insert(nil, "schema", [:foo, :bar], query, {:raise, [], []}, [:foo])
+
+    assert query == ~s{INSERT INTO [schema] ([foo],[bar]) OUTPUT INSERTED.[foo] SELECT 3, s0.[bar] FROM [schema] AS s0}
+  end
+
   test "update" do
     query = update(nil, "schema", [:id], [:x, :y], [])
     assert query == ~s{UPDATE [schema] SET [id] = @1 WHERE [x] = @2 AND [y] = @3}