Fix map update source field caching

lib/ecto/query/planner.ex

@@ -1785,7 +1785,7 @@ defmodule Ecto.Query.Planner do
 
     {fields, preprocess, from} =
       case from do
-        {from_expr, from_source, from_fields} ->
+        {from_expr, from_source, from_fields, _drop} ->
           {assoc_exprs, assoc_fields} = collect_assocs([], [], query, tag, from_take, assocs)
           fields = from_fields ++ Enum.reverse(assoc_fields, Enum.reverse(fields))
           preprocess = [from_expr | Enum.reverse(assoc_exprs)]
@@ -1831,13 +1831,14 @@ defmodule Ecto.Query.Planner do
        ) do
     case collect_fields(left, fields, from, query, take, keep_literals?, %{}) do
       {{:source, :from}, fields, left_from} ->
-        {right, right_fields, _} =
+        {right, right_fields, right_from} =
           collect_fields(right, [], left_from, query, take, keep_literals?, %{})
 
-        {from_expr, from_source, from_fields} = left_from
+        {from_expr, from_source, from_fields, drop} = right_from
 
         from =
-          {{:merge, from_expr, right}, from_source, from_fields ++ Enum.reverse(right_fields)}
+          {{:merge, from_expr, right}, from_source, from_fields ++ Enum.reverse(right_fields),
+           drop}
 
         {{:source, :from}, fields, from}
 
@@ -1851,10 +1852,30 @@ defmodule Ecto.Query.Planner do
 
   defp collect_fields({:&, _, [0]}, fields, :none, query, take, _keep_literals?, drop) do
     {expr, taken} = source_take!(:select, query, take, 0, 0, drop)
-    {{:source, :from}, fields, {{:source, :from}, expr, taken}}
+    {{:source, :from}, fields, {{:source, :from}, expr, taken, drop}}
   end
 
-  defp collect_fields({:&, _, [0]}, fields, from, _query, _take, _keep_literals?, _drop) do
+  defp collect_fields(
+         {:&, _, [0]},
+         fields,
+         {from_expr, _, _, cached_drop} = from,
+         query,
+         take,
+         _keep_literals?,
+         drop
+       ) do
+    # All references to the from binding share this source, so a field can only
+    # be dropped when every full-source reference overwrites it.
+    drop = Map.take(cached_drop, Map.keys(drop))
+
+    from =
+      if drop == cached_drop do
+        from
+      else
+        {from_source, from_fields} = source_take!(:select, query, take, 0, 0, drop)
+        {from_expr, from_source, from_fields, drop}
+      end
+
     {{:source, :from}, fields, from}
   end
 

test/ecto/query/planner_test.exs

@@ -2671,6 +2671,81 @@ defmodule Ecto.Query.PlannerTest do
                select_fields([:id, :posted, :uuid, :crazy_comment, :post_id, :crazy_post_id], 1)
   end
 
+  test "normalize: map update does not drop fields from another full source reference" do
+    fields =
+      select_fields(
+        [
+          :id,
+          :post_title,
+          :text,
+          :code,
+          :posted,
+          :visits,
+          :links,
+          :preferences,
+          :status,
+          :parameterized_map,
+          :meta,
+          :metas
+        ],
+        0
+      )
+
+    query = Post |> select([p], {%{p | title: nil}, p}) |> normalize()
+    assert query.select.fields == fields
+
+    query = Post |> select([p], {p, %{p | title: nil}}) |> normalize()
+    assert query.select.fields == fields
+  end
+
+  test "normalize: map updates only drop fields overwritten by every source reference" do
+    query =
+      Post
+      |> select([p], {%{p | title: nil}, %{p | title: nil, posted: nil}})
+      |> normalize()
+
+    assert query.select.fields ==
+             select_fields(
+               [
+                 :id,
+                 :text,
+                 :code,
+                 :posted,
+                 :visits,
+                 :links,
+                 :preferences,
+                 :status,
+                 :parameterized_map,
+                 :meta,
+                 :metas
+               ],
+               0
+             )
+  end
+
+  test "normalize: struct update does not drop fields from another full source reference" do
+    query = Post |> select([p], {%Post{p | title: nil}, p}) |> normalize()
+
+    assert query.select.fields ==
+             select_fields(
+               [
+                 :id,
+                 :post_title,
+                 :text,
+                 :code,
+                 :posted,
+                 :visits,
+                 :links,
+                 :preferences,
+                 :status,
+                 :parameterized_map,
+                 :meta,
+                 :metas
+               ],
+               0
+             )
+  end
+
   test "normalize: select single dynamic value interpolated at root level" do
     ref = dynamic([p], p.title)