Propagate opts configured in prepare_query/3 to preloads

lib/ecto/repo.ex

@@ -644,11 +644,17 @@ defmodule Ecto.Repo do
         def preload(struct_or_structs_or_nil, preloads, opts \\ []) do
           repo = get_dynamic_repo()
 
+          {adapter_meta, opts} =
+            _tuplet =
+            Ecto.Repo.Supervisor.tuplet(repo, prepare_opts(:preload, opts))
+
+          {_query, opts} = repo.prepare_query(:preload, struct_or_structs_or_nil, opts)
+
           Ecto.Repo.Preloader.preload(
             struct_or_structs_or_nil,
             repo,
             preloads,
-            Ecto.Repo.Supervisor.tuplet(repo, prepare_opts(:preload, opts))
+            {adapter_meta, opts}
           )
         end
 

lib/ecto/repo/queryable.ex

@@ -213,7 +213,7 @@ defmodule Ecto.Repo.Queryable do
 
   ## Helpers
 
-  defp execute(operation, name, query, {adapter_meta, opts} = tuplet) do
+  defp execute(operation, name, query, {adapter_meta, opts} = _tuplet) do
     %{adapter: adapter, cache: cache, repo: repo} = adapter_meta
 
     {query, opts} = repo.prepare_query(operation, query, opts)
@@ -244,7 +244,14 @@ defmodule Ecto.Repo.Queryable do
         {count,
          rows
          |> Ecto.Repo.Assoc.query(assocs, sources, preprocessor)
-         |> Ecto.Repo.Preloader.query(name, preloads, take, assocs, postprocessor, tuplet)}
+         |> Ecto.Repo.Preloader.query(
+           name,
+           preloads,
+           take,
+           assocs,
+           postprocessor,
+           {adapter_meta, opts}
+         )}
     end
   end
 

test/ecto/repo_test.exs

@@ -182,7 +182,7 @@
   end
 
   defmodule MySchemaOneField do
-     use Ecto.Schema
+    use Ecto.Schema
 
     @primary_key false
     schema "my_schema" do
@@ -2211,6 +2211,13 @@
 
       def prepare_query(op, query, opts) do
         send(self(), {op, query, opts})
+
+        opts =
+          case Keyword.fetch(opts, :for_on_preloader_spawn) do
+            {:ok, fun} -> [{:on_preloader_spawn, fun} | opts]
+            _ -> opts
+          end
+
         {%{query | prefix: "rewritten"}, opts}
       end
     end
@@ -2253,13 +2260,13 @@
       assert_received {:stream, %{prefix: "rewritten"}}
     end
 
     test "preload" do
       PrepareRepo.preload(%MySchemaWithAssoc{parent_id: 1}, :parent, hello: :world)
       assert_received {:all, query, [ecto_query: :preload, hello: :world]}
       assert query.from.source == {"my_parent", Ecto.RepoTest.MyParent}
     end
 
     test "preload with :on_preloader_spawn" do
       test_process = self()
       fun = fn -> send(test_process, {:callback_ran, self()}) end
 
@@ -2270,6 +2277,32 @@
       assert_received {:callback_ran, pid2} when pid2 != self()
       assert pid1 != pid2
     end
+
+    test "preload with :on_preloader_spawn in prepare_query/3 callback" do
+      test_process = self()
+      fun = fn -> send(test_process, {:callback_ran, self()}) end
+
+      %MySchemaWithMultiAssoc{parent_id: 1, mother_id: 2}
+      |> PrepareRepo.preload([:parent, :mother], for_on_preloader_spawn: fun)
+
+      assert_received {:callback_ran, pid1} when pid1 != self()
+      assert_received {:callback_ran, pid2} when pid2 != self()
+      assert pid1 != pid2
+    end
+
+    test "all with preload with :on_preloader_spawn in prepare_query/3 callback" do
+      test_process = self()
+      fun = fn -> send(test_process, {:callback_ran, self()}) end
+
+      from(p in MyParent,
+        preload: [:parent, :mother]
+      )
+      |> PrepareRepo.all(for_on_preloader_spawn: fun)
+
+      assert_received {:callback_ran, pid1} when pid1 != self()
+      assert_received {:callback_ran, pid2} when pid2 != self()
+      assert pid1 != pid2
+    end
   end
 
   describe "prepare_transaction" do