Fix subqueries in interpolated join wheres

lib/ecto/query/inspect.ex

@@ -1,7 +1,7 @@
 import Inspect.Algebra
 import Kernel, except: [to_string: 1]
 
-alias Ecto.Query.{DynamicExpr, JoinExpr, QueryExpr, WithExpr, LimitExpr}
+alias Ecto.Query.{DynamicExpr, JoinExpr, WithExpr, LimitExpr}
 
 defimpl Inspect, for: Ecto.Query.DynamicExpr do
   def inspect(%DynamicExpr{binding: binding} = dynamic, opts) do
@@ -191,8 +191,10 @@ defimpl Inspect, for: Ecto.Query do
     [{join_qual(qual), string}] ++ kw_as_and_prefix(join) ++ [on: expr(on, names)]
   end
 
-  defp maybe_on(%QueryExpr{expr: true}, _names), do: []
-  defp maybe_on(%QueryExpr{} = on, names), do: [on: expr(on, names)]
+  # The `on` may be a QueryExpr or, while an interpolated join query
+  # is being planned, a BooleanExpr
+  defp maybe_on(%{expr: true}, _names), do: []
+  defp maybe_on(%{} = on, names), do: [on: expr(on, names)]
 
   defp preloads([]), do: []
   defp preloads(preloads), do: [preload: inspect(preloads)]

lib/ecto/query/planner.ex

@@ -27,11 +27,11 @@ defmodule Ecto.Query.Planner do
   in order to keep proper binding order.
   """
   def query_to_joins(qual, source, %{wheres: wheres, joins: joins}, position) do
-    on = %QueryExpr{file: __ENV__.file, line: __ENV__.line, expr: true, params: []}
+    on = %BooleanExpr{op: :and, file: __ENV__.file, line: __ENV__.line, expr: true, params: []}
 
     on =
-      Enum.reduce(wheres, on, fn %BooleanExpr{op: op, expr: expr, params: params}, acc ->
-        merge_expr_and_params(op, acc, expr, params)
+      Enum.reduce(wheres, on, fn %BooleanExpr{op: op} = expr, acc ->
+        merge_expr_and_params(op, acc, expr)
       end)
 
     join = %JoinExpr{qual: qual, source: source, file: __ENV__.file, line: __ENV__.line, on: on}
@@ -47,14 +47,71 @@ defmodule Ecto.Query.Planner do
     end
   end
 
+  defp merge_expr_and_params(
+         op,
+         %BooleanExpr{expr: left_expr, params: left_params, subqueries: left_subqueries} = struct,
+         %BooleanExpr{
+           expr: right_expr,
+           params: right_params,
+           subqueries: right_subqueries
+         }
+       ) do
+    merge_expr_and_params(
+      op,
+      struct,
+      left_expr,
+      left_params,
+      left_subqueries,
+      right_expr,
+      right_params,
+      right_subqueries
+    )
+  end
+
   defp merge_expr_and_params(
          op,
          %QueryExpr{expr: left_expr, params: left_params} = struct,
+         %BooleanExpr{expr: right_expr, params: right_params, subqueries: []}
+       ) do
+    merge_expr_and_params(op, struct, left_expr, left_params, [], right_expr, right_params, [])
+  end
+
+  defp merge_expr_and_params(
+         op,
+         struct,
+         left_expr,
+         left_params,
+         left_subqueries,
          right_expr,
-         right_params
+         right_params,
+         right_subqueries
        ) do
-    right_expr = Ecto.Query.Builder.bump_interpolations(right_expr, left_params)
-    %{struct | expr: merge_expr(op, left_expr, right_expr), params: left_params ++ right_params}
+    right_expr =
+      right_expr
+      |> Ecto.Query.Builder.bump_interpolations(left_params)
+      |> Ecto.Query.Builder.bump_subqueries(left_subqueries)
+
+    right_params = bump_subquery_params(right_params, left_subqueries)
+
+    struct = %{
+      struct
+      | expr: merge_expr(op, left_expr, right_expr),
+        params: left_params ++ right_params
+    }
+
+    case left_subqueries ++ right_subqueries do
+      [] -> struct
+      subqueries -> %{struct | subqueries: subqueries}
+    end
+  end
+
+  defp bump_subquery_params(params, subqueries) do
+    len = length(subqueries)
+
+    Enum.map(params, fn
+      {:subquery, counter} -> {:subquery, len + counter}
+      other -> other
+    end)
   end
 
   defp merge_expr(_op, left, true), do: left
@@ -227,6 +284,7 @@ defmodule Ecto.Query.Planner do
 
     query
     |> plan_assocs()
+    |> plan_join_subqueries(plan_subquery)
     |> plan_combinations(adapter, cte_names)
     |> plan_expr_subqueries(:wheres, plan_subquery)
     |> plan_expr_subqueries(:havings, plan_subquery)
@@ -746,8 +804,18 @@ defmodule Ecto.Query.Planner do
     {joins, sources, tail_sources}
   end
 
-  defp attach_on([%{on: on} = h | t], %{expr: expr, params: params}) do
-    [%{h | on: merge_expr_and_params(:and, on, expr, params)} | t]
+  defp attach_on(joins, %QueryExpr{expr: expr, file: file, line: line, params: params}) do
+    attach_on(joins, %BooleanExpr{
+      op: :and,
+      expr: expr,
+      file: file,
+      line: line,
+      params: params
+    })
+  end
+
+  defp attach_on([%{on: on} = h | t], %BooleanExpr{} = expr) do
+    [%{h | on: merge_expr_and_params(:and, on, expr)} | t]
   end
 
   defp rewrite_prefix(expr, nil), do: expr
@@ -879,6 +947,19 @@ defmodule Ecto.Query.Planner do
     query
   end
 
+  defp plan_join_subqueries(query, fun) do
+    joins =
+      Enum.map(query.joins, fn
+        %{on: %BooleanExpr{subqueries: [_ | _] = subqueries} = on} = join ->
+          %{join | on: %{on | subqueries: Enum.map(subqueries, fun)}}
+
+        join ->
+          join
+      end)
+
+    %{query | joins: joins}
+  end
+
   defp plan_expr_subquery(query, key, fun) do
     with %{^key => %{subqueries: [_ | _] = subqueries} = expr} <- query do
       %{query | key => %{expr | subqueries: Enum.map(subqueries, fun)}}
@@ -952,7 +1033,7 @@ defmodule Ecto.Query.Planner do
           {params, join_cacheable?} = cast_and_merge_params(:join, query, join, params, adapter)
           {params, on_cacheable?} = cast_and_merge_params(:join, query, on, params, adapter)
 
-          {{qual, key, on.expr, hints},
+          {{qual, key, expr_to_cache(on), hints},
            {params, cacheable? and join_cacheable? and on_cacheable? and key != :nocache}}
       end)
 
@@ -1422,7 +1503,7 @@ defmodule Ecto.Query.Planner do
     Enum.map_reduce(exprs, counter, fn join, acc ->
       {source, acc} = prewalk_source(join.source, :join, query, join, acc, adapter)
       {on, acc} = prewalk(:join, query, join.on, acc, adapter)
-      {%{join | on: on, source: source, params: nil}, acc}
+      {%{join | on: on_to_query_expr(on), source: source, params: nil}, acc}
     end)
   end
 
@@ -1453,6 +1534,15 @@ defmodule Ecto.Query.Planner do
     {Enum.reverse(combinations), counter}
   end
 
+  # Interpolated join queries carry their `on` as a BooleanExpr during
+  # planning, as it may hold subqueries. Once subqueries are inlined by
+  # prewalk, convert it back to the QueryExpr adapters expect.
+  defp on_to_query_expr(%BooleanExpr{expr: expr, file: file, line: line, params: params}) do
+    %QueryExpr{expr: expr, file: file, line: line, params: params}
+  end
+
+  defp on_to_query_expr(on), do: on
+
   defp validate_json_path!([path_field | rest], field, {:parameterized, {Ecto.Embedded, embed}})
        when is_binary(path_field) or is_integer(path_field) do
     case embed do

test/ecto/query/planner_test.exs

@@ -648,6 +648,71 @@ defmodule Ecto.Query.PlannerTest do
     assert key == :nocache
   end
 
+  test "plan: interpolated join query with a subquery in where" do
+    subquery = from(s in "subposts", select: s.id)
+    join_query = from(p in "posts", where: p.id in subquery(subquery))
+    query = from(p in Post, join: p2 in ^join_query, on: true)
+
+    {planned, _, _, _} = plan(query)
+
+    assert [
+             %{
+               on: %{
+                 expr: {:in, _, [_, {:subquery, 0}]},
+                 subqueries: [%Ecto.SubQuery{}]
+               }
+             }
+           ] = planned.joins
+
+    # adapters pattern match on `%JoinExpr{on: %QueryExpr{}}`, so the
+    # BooleanExpr used during planning must not leak into the normalized query
+    assert [%{on: %Ecto.Query.QueryExpr{expr: {:in, _, [_, %Ecto.SubQuery{}]}}}] =
+             normalize(query).joins
+  end
+
+  test "plan: join cache includes subqueries from interpolated wheres" do
+    first_subquery = from(s in "first_subposts", select: s.id)
+    second_subquery = from(s in "second_subposts", select: s.id)
+
+    first_query =
+      from(p in Post,
+        join: p2 in ^from(p in "posts", where: p.id in subquery(first_subquery)),
+        on: true
+      )
+
+    second_query =
+      from(p in Post,
+        join: p2 in ^from(p in "posts", where: p.id in subquery(second_subquery)),
+        on: true
+      )
+
+    {_, _, _, first_key} = plan(first_query)
+    {_, _, _, second_key} = plan(second_query)
+
+    refute first_key == second_key
+  end
+
+  test "plan: merges subqueries from interpolated join wheres" do
+    first_subquery = from(s in "first_subposts", where: s.id == ^1, select: s.id)
+    second_subquery = from(s in "second_subposts", where: s.id == ^2, select: s.id)
+
+    join_query =
+      from(p in "posts",
+        where: p.id in subquery(first_subquery),
+        or_where: p.id in subquery(second_subquery)
+      )
+
+    {query, cast_params, dump_params, _} =
+      from(p in Post, join: p2 in ^join_query, on: true) |> plan()
+
+    assert cast_params == [1, 2]
+    assert dump_params == [1, 2]
+
+    assert [%{on: %{expr: {:or, _, [_, _]}, subqueries: [first, second]}}] = query.joins
+    assert %Ecto.SubQuery{query: %{from: %{source: {"first_subposts", nil}}}} = first
+    assert %Ecto.SubQuery{query: %{from: %{source: {"second_subposts", nil}}}} = second
+  end
+
   test "plan: normalizes prefixes" do
     # No schema prefix in from
     {query, _, _, _} = from(Comment, select: 1) |> plan()