Allow {fragment, Schema} sources in joins

integration_test/cases/joins.exs

@@ -5,6 +5,7 @@ defmodule Ecto.Integration.JoinsTest do
   import Ecto.Query
 
   alias Ecto.Integration.Post
+  alias Ecto.Integration.Barebone
   alias Ecto.Integration.Comment
   alias Ecto.Integration.Permalink
   alias Ecto.Integration.User
@@ -162,6 +163,16 @@ defmodule Ecto.Integration.JoinsTest do
     assert p2.permalink.id == plid1
   end
 
+  test "joins with fragment source mapped to schema" do
+    query =
+      from f1  in {fragment("select 1 as num"), Barebone},
+        join: f2  in {fragment("select 1 as visits"), Post},
+        on: f1.num == f2.visits,
+        select: {f1, struct(f2, [:visits])}
+
+    assert {%Barebone{num: 1} = b1, %Post{visits: 1} = b2} = TestRepo.one(query)
+  end
+
   ## Associations joins
 
   test "has_many association join" do

lib/ecto/query/builder/join.ex

@@ -81,6 +81,18 @@ defmodule Ecto.Query.Builder.Join do
     end
   end
 
+  def escape({{:fragment, _, _} = fragment, schema} = join, vars, env) do
+    {_, expr, _, _, params} = escape(fragment, vars, env)
+
+    case Macro.expand(schema, env) do
+      schema when is_atom(schema) ->
+        {:_, {expr, schema}, nil, nil, params}
+
+      _ ->
+        Builder.error!("malformed join `#{Macro.to_string(join)}` in query expression")
+    end
+  end
+
   def escape({:assoc, _, [{var, _, context}, field]}, vars, _env)
       when is_atom(var) and is_atom(context) do
     ensure_field!(field)

test/ecto/query/planner_test.exs

@@ -949,13 +949,26 @@ defmodule Ecto.Query.PlannerTest do
   end
 
   test "plan: tuple source with fragment" do
-    {query, cast_params, dump_params, cache_key} =
-      plan(from {fragment("? as num", ^0), Barebone})
+    query =
+      from f1 in {fragment("? as num", ^0), Barebone},
+        join: f2 in {fragment("? as visits", ^0), Post},
+        on: f1.num == f2.visits,
+        select: {f1, f2}
+
+    {query, cast_params, dump_params, cache_key} = plan(query)
 
-    assert {{{:fragment, [], _}, Barebone, nil}} = query.sources
-    assert cast_params == [0]
-    assert dump_params == [0]
-    assert [:all, {:from, {{:fragment, _, _}, Barebone, _, _}, []}] = cache_key
+    assert {from_source, join_source} = query.sources
+    assert {{:fragment, [], _}, Barebone, nil} = from_source
+    assert {{:fragment, [], _}, Post, nil} = join_source
+    assert cast_params == [0, 0]
+    assert dump_params == [0, 0]
+
+    assert [
+             :all,
+             {:join, [{:inner, {{:fragment, _, _}, Post, _, _}, {:==, _, _}, []}]},
+             {:from, {{:fragment, _, _}, Barebone, _, _}, []},
+             {:select, {:{}, [], [{:&, [], [0]}, {:&, [], [1]}]}}
+           ] = cache_key
   end
 
   test "plan: tuple source with fragment and take" do
@@ -2609,13 +2622,23 @@ defmodule Ecto.Query.PlannerTest do
   end
 
   test "normalize: tuple source with fragment" do
-    {query, _, _, select} =
-      normalize_with_params(from {fragment("? as num", ^0), Barebone})
+    query =
+      from f1 in {fragment("? as num", ^0), Barebone},
+        join: f2 in {fragment("? as num", ^0), Barebone},
+        on: f1.num == f2.num,
+        select: {f1, f2}
 
-    %{from: {_, {:source, {{:fragment, _, _}, Barebone}, nil, types}}} = select
-    assert types == [num: :integer]
+    {query, _, _, select} = normalize_with_params(query)
+
+    %{from: {_, {:source, {{:fragment, _, _}, Barebone}, nil, from_types}}} = select
+    assert from_types == [num: :integer]
     assert {{:fragment, _, _}, Barebone} = query.from.source
-    assert query.select.fields == [{{:., [writable: :always], [{:&, [], [0]}, :num]}, [], []}]
+    assert [%{source: {{:fragment, _, _}, Barebone}}] = query.joins
+
+    assert query.select.fields == [
+             {{:., [writable: :always], [{:&, [], [0]}, :num]}, [], []},
+             {{:., [writable: :always], [{:&, [], [1]}, :num]}, [], []}
+           ]
   end
 
   test "normalize: tuple source with fragment and take" do