Map insert_all source fields through the destination schema (#4795)

This PR maps selected fields through the destination schema. It keeps headers aligned with the source select list. This addresses the cases where destination writeable: true and renamed columns were not respected. It also hardens checks against duplicate projections and other unsupported cases. Tests cover mapped and read-only fields, map updates, and schema and non-schema sources

Fixes #4794

Co-authored-by: José Valim <jose.valim@gmail.com>

Author: lukaszsamson

lib/ecto/repo/schema.ex

@@ -175,48 +175,23 @@ defmodule Ecto.Repo.Schema do
 
     header =
       case query.select do
-        %Ecto.Query.SelectExpr{expr: {:%{}, [], [{:|, _, [{:&, _, [ix]}, args]}]}, fields: fields} ->
-          {updated_fields, updated_set} =
-            Enum.map_reduce(args, MapSet.new(), fn {field, _}, set ->
-              dumped_field = insert_all_select_dump!(field, dumper)
-              {dumped_field, MapSet.put(set, dumped_field)}
-            end)
+        %Ecto.Query.SelectExpr{expr: {:%{}, [], [{:|, _, [{:&, _, [ix]}, args]}]}} ->
+          updated_fields =
+            Enum.map(args, fn {field, _} -> insert_all_select_dump!(field, dumper) end)
 
-          unchanged_fields =
-            for {{:., _, [{:&, _, [^ix]}, field]}, [], []} = expr <- fields,
-                not MapSet.member?(updated_set, field),
-                do: insert_all_select_dump!(expr)
-
-          unchanged_fields ++ updated_fields
+          insert_all_source_fields(query, ix, args, dumper) ++ updated_fields
 
         %Ecto.Query.SelectExpr{expr: {:%{}, _ctx, args}} ->
           Enum.map(args, fn {field, _} -> insert_all_select_dump!(field, dumper) end)
 
         %Ecto.Query.SelectExpr{take: %{^ix => {_fun, fields}}} ->
           Enum.map(fields, &insert_all_select_dump!(&1, dumper))
 
-        %Ecto.Query.SelectExpr{expr: {:&, _, [_ix]}, fields: fields} ->
-          Enum.map(fields, &insert_all_select_dump!(&1))
+        %Ecto.Query.SelectExpr{expr: {:&, _, [ix]}} ->
+          insert_all_source_fields(query, ix, [], dumper)
 
         _ ->
-          raise ArgumentError, """
-          cannot generate a fields list for insert_all from the given source query:
-
-            #{inspect(query)}
-
-          The select clause must be one of the following:
-
-            * A single `map/2` or several `map/2` expressions combined with `select_merge`
-            * A single `struct/2` or several `struct/2` expressions combined with `select_merge`
-            * A source such as `p` in the query `from p in Post`
-            * A single literal map or several literal maps combined with `select_merge`. If
-              combining several literal maps, there cannot be any query interpolations
-              except in the last `select_merge`. Consider using `Ecto.Query.exclude/2`
-              to rebuild the select expression from scratch if you need multiple `select_merge`
-              statements with interpolations
-
-          All keys must exist in the schema that is being inserted into
-          """
+          insert_all_select_error!(query)
       end
 
     counter = fn -> length(dump_params) end
@@ -336,27 +311,99 @@ defmodule Ecto.Repo.Schema do
     {rows, Enum.reverse(cast_params), counter}
   end
 
-  defp insert_all_select_dump!({{:., dot_meta, [{:&, _, [_]}, field]}, [], []}) do
-    if dot_meta[:writable] == :never do
-      raise ArgumentError, "cannot select unwritable field `#{inspect(field)}` for insert_all"
-    else
-      field
+  defp insert_all_source_fields(query, ix, updates, dumper) do
+    fields = query.select.fields
+    count = length(fields) - length(updates)
+    if count < 0, do: insert_all_select_error!(query)
+    fields = Enum.take(fields, count)
+    updates = Map.new(updates)
+
+    # Schema fields are logical names; SELECT fields contain physical column names.
+    case elem(query.sources, ix) do
+      {_, schema, _} when is_atom(schema) and not is_nil(schema) ->
+        selected =
+          case query.select.take do
+            %{^ix => {_, selected}} -> selected
+            _ -> schema.__schema__(:query_fields)
+          end
+
+        source_dumper = schema.__schema__(:dump)
+
+        {header, leftover} =
+          for field <- selected, is_atom(field), not is_map_key(updates, field), reduce: {[], fields} do
+            {header, fields} ->
+              source =
+                case source_dumper do
+                  %{^field => {source, _, _}} -> source
+                  _ -> field
+                end
+
+              case fields do
+                [{{:., _, [{:&, _, [^ix]}, ^source]}, [], []} | fields] ->
+                  field = insert_all_select_dump!(if(dumper, do: field, else: source), dumper)
+                  {[field | header], fields}
+
+                _ ->
+                  insert_all_select_error!(query)
+              end
+          end
+
+        if leftover != [], do: insert_all_select_error!(query)
+        Enum.reverse(header)
+
+      _ ->
+        Enum.map(fields, fn
+          {{:., _, [{:&, _, [^ix]}, field]}, [], []} ->
+            if is_map_key(updates, field), do: insert_all_select_error!(query)
+            insert_all_select_dump!(field, dumper)
+
+          _ ->
+            insert_all_select_error!(query)
+        end)
     end
   end
 
+  defp insert_all_select_error!(query) do
+    raise ArgumentError, """
+    cannot generate a fields list for insert_all from the given source query:
+
+      #{inspect(query)}
+
+    The select clause must be one of the following:
+
+      * A single `map/2` or several `map/2` expressions combined with `select_merge`
+      * A single `struct/2` or several `struct/2` expressions combined with `select_merge`
+      * A source such as `p` in the query `from p in Post`
+      * A single literal map or several literal maps combined with `select_merge`. If
+        combining several literal maps, there cannot be any query interpolations
+        except in the last `select_merge`. Consider using `Ecto.Query.exclude/2`
+        to rebuild the select expression from scratch if you need multiple `select_merge`
+        statements with interpolations
+
+    All keys must exist in the schema that is being inserted into
+    """
+  end
+
   defp insert_all_select_dump!(field, dumper) when is_atom(field) do
     case dumper do
       %{^field => {source, _, writable}} when writable != :never ->
         source
 
-      %{} ->
+      %{^field => {_, _, :never}} ->
         raise ArgumentError, "cannot select unwritable field `#{inspect(field)}` for insert_all"
 
+      %{} ->
+        raise ArgumentError, "cannot select unknown field `#{inspect(field)}` for insert_all"
+
       nil ->
         field
     end
   end
 
+  defp insert_all_select_dump!(field, _dumper) do
+    raise ArgumentError, "cannot select non-atom field `#{inspect(field)}` for insert_all"
+  end
+
   defp autogenerate_id(nil, fields, header, _adapter) do
     {fields, header}
   end

test/ecto/repo_test.exs

@@ -211,6 +211,50 @@ defmodule Ecto.RepoTest do
     end
   end
 
+  defmodule InsertSelectRenamed do
+    use Ecto.Schema
+
+    @primary_key false
+    schema "insert_select_renamed" do
+      field :name, :string, source: :renamed_name
+      field :value, :string
+    end
+  end
+
+  defmodule InsertSelectReadOnly do
+    use Ecto.Schema
+
+    @primary_key false
+    schema "insert_select_read_only" do
+      field :name, :string, source: :source_name, writable: :never
+      field :value, :string
+    end
+  end
+
+  defmodule InsertSelectDisjointSource do
+    use Ecto.Schema
+
+    @primary_key false
+    schema "insert_select_disjoint_source" do
+      field :b, :string, source: :src_b
+      field :c, :string, source: :src_c
+      field :a, :string, source: :src_a
+      field :only_source, :string
+    end
+  end
+
+  defmodule InsertSelectDisjointDestination do
+    use Ecto.Schema
+
+    @primary_key false
+    schema "insert_select_disjoint_destination" do
+      field :only_destination, :string
+      field :a, :string
+      field :c, :string, source: :dst_c
+      field :b, :string, source: :dst_b
+    end
+  end
+
   test "defines child_spec/1" do
     assert TestRepo.child_spec([]) == %{
              id: TestRepo,
@@ -765,6 +809,167 @@ defmodule Ecto.RepoTest do
       assert header == [:id, :x, :yyy, :z, :array, :map]
     end
 
+    test "maps read-only source fields through the writable destination schema" do
+      query = from s in InsertSelectReadOnly, select: s
+      TestRepo.insert_all(InsertSelectRenamed, query)
+
+      assert_received {:insert_all, %{header: [:renamed_name, :value]},
+                       {%Ecto.Query{select: %{fields: fields}}, _params}}
+
+      assert [
+               {{:., _, [{:&, _, [0]}, :source_name]}, [], []},
+               {{:., _, [{:&, _, [0]}, :value]}, [], []}
+             ] = fields
+    end
+
+    test "maps fragment columns through the destination schema" do
+      query =
+        from f in fragment("select 1 as name, 2 as value", columns: [:name, :value]), select: f
+
+      TestRepo.insert_all(InsertSelectRenamed, query)
+
+      assert_received {:insert_all, %{header: [:renamed_name, :value]}, {%Ecto.Query{}, _params}}
+    end
+
+    test "maps subquery fields through the destination schema" do
+      inner = from s in InsertSelectReadOnly, select: %{name: s.name, value: s.value}
+      query = from s in subquery(inner), select: s
+      TestRepo.insert_all(InsertSelectRenamed, query)
+
+      assert_received {:insert_all, %{header: [:renamed_name, :value]}, {%Ecto.Query{}, _params}}
+    end
+
+    test "rejects a map update that repeats a subquery field" do
+      inner = from s in InsertSelectReadOnly, select: %{name: s.name, value: s.value}
+      query = from s in subquery(inner), select: %{s | value: "new"}
+
+      # This query shape still projects the overwritten field from the subquery.
+      # Reject the duplicate projection; this does not add support for subquery map updates.
+      assert_raise ArgumentError,
+                   ~r/cannot generate a fields list for insert_all from the given source query:/,
+                   fn -> TestRepo.insert_all(InsertSelectRenamed, query) end
+    end
+
+    test "maps values fields through the destination schema" do
+      query =
+        from v in values([%{name: "n", value: "v"}], %{name: :string, value: :string}),
+          select: %{v | value: "new"}
+
+      TestRepo.insert_all(InsertSelectRenamed, query)
+
+      assert_received {:insert_all, %{header: [:renamed_name, :value]}, {%Ecto.Query{}, _params}}
+    end
+
+    test "maps a joined binding through the destination schema" do
+      query = from x in "other", join: s in InsertSelectReadOnly, on: true, select: s
+      TestRepo.insert_all(InsertSelectRenamed, query)
+
+      assert_received {:insert_all, %{header: [:renamed_name, :value]}, {%Ecto.Query{}, _params}}
+    end
+
+    test "keeps source columns when the destination has no schema" do
+      query = from s in InsertSelectReadOnly, select: s
+      TestRepo.insert_all("insert_select_renamed", query)
+
+      assert_received {:insert_all, %{header: [:source_name, :value]}, {%Ecto.Query{}, _params}}
+    end
+
+    test "rejects full source fields that are unwritable in the destination" do
+      query = from s in InsertSelectRenamed, select: s
+
+      assert_raise ArgumentError,
+                   "cannot select unwritable field `:name` for insert_all",
+                   fn -> TestRepo.insert_all(InsertSelectReadOnly, query) end
+    end
+
+    test "maps unchanged map update fields through the destination schema" do
+      query = from s in InsertSelectReadOnly, select: %{s | value: s.value}
+      TestRepo.insert_all(InsertSelectRenamed, query)
+
+      assert_received {:insert_all, %{header: [:renamed_name, :value]},
+                       {%Ecto.Query{select: %{fields: fields}}, _params}}
+
+      assert [
+               {{:., _, [{:&, _, [0]}, :source_name]}, [], []},
+               {{:., _, [{:&, _, [0]}, :value]}, [], []}
+             ] = fields
+    end
+
+    test "does not include an overwritten source field twice when columns differ" do
+      query = from s in InsertSelectReadOnly, select: %{s | name: s.name}
+      TestRepo.insert_all(InsertSelectRenamed, query)
+
+      assert_received {:insert_all, %{header: [:value, :renamed_name]}, {%Ecto.Query{}, _params}}
+    end
+
+    test "rejects unchanged map update fields that are unwritable in the destination" do
+      query = from s in InsertSelectRenamed, select: %{s | value: s.value}
+
+      assert_raise ArgumentError,
+                   "cannot select unwritable field `:name` for insert_all",
+                   fn -> TestRepo.insert_all(InsertSelectReadOnly, query) end
+    end
+
+    test "rejects map updates whose values expand to multiple select fields" do
+      query = from s in InsertSelectReadOnly, select: %{s | value: {s.name, s.value}}
+
+      assert_raise ArgumentError,
+                   ~r/cannot generate a fields list for insert_all from the given source query:/,
+                   fn -> TestRepo.insert_all(InsertSelectRenamed, query) end
+    end
+
+    test "maps a reordered source subset independently of destination field order" do
+      query =
+        from s in InsertSelectDisjointSource,
+          select: %{map(s, [:b, :c, :a]) | a: fragment("'x'")}
+
+      TestRepo.insert_all(InsertSelectDisjointDestination, query)
+
+      assert_received {:insert_all, %{header: [:dst_b, :dst_c, :a]},
+                       {%Ecto.Query{select: %{fields: fields}}, _params}}
+
+      assert [
+               {{:., _, [{:&, _, [0]}, :src_b]}, [], []},
+               {{:., _, [{:&, _, [0]}, :src_c]}, [], []},
+               {:fragment, _, _}
+             ] = fields
+    end
+
+    test "allows destination-only map updates" do
+      query =
+        from s in InsertSelectDisjointSource,
+          select: %{map(s, [:b, :c]) | a: s.c, only_destination: s.b}
+
+      TestRepo.insert_all(InsertSelectDisjointDestination, query)
+
+      assert_received {:insert_all, %{header: [:dst_b, :dst_c, :a, :only_destination]},
+                       {%Ecto.Query{}, _params}}
+    end
+
+    test "reports source-only fields" do
+      query = from s in InsertSelectDisjointSource, select: s
+
+      assert_raise ArgumentError,
+                   "cannot select unknown field `:only_source` for insert_all",
+                   fn -> TestRepo.insert_all(InsertSelectDisjointDestination, query) end
+    end
+
+    test "rejects whole schemaless bindings with a select error" do
+      query = from s in "insert_select_source", select: s
+
+      assert_raise ArgumentError,
+                   ~r/cannot generate a fields list for insert_all from the given source query:/,
+                   fn -> TestRepo.insert_all(InsertSelectRenamed, query) end
+    end
+
+    test "rejects non-atom insert select keys with a clear error" do
+      query = from s in InsertSelectReadOnly, select: %{"name" => s.name}
+
+      assert_raise ArgumentError,
+                   "cannot select non-atom field `\"name\"` for insert_all",
+                   fn -> TestRepo.insert_all(InsertSelectRenamed, query) end
+    end
+
     test "takes query selecting on source with join" do
       query = from p in MyParent, join: a in MySchemaWithAssoc, on: true, select: a
       TestRepo.insert_all(MySchemaWithAssoc, query)