diff --git a/lib/ecto/query/builder/join.ex b/lib/ecto/query/builder/join.ex
index 5fd160c97a..5eebaf46fa 100644
--- a/lib/ecto/query/builder/join.ex
+++ b/lib/ecto/query/builder/join.ex
@@ -4,7 +4,7 @@ defmodule Ecto.Query.Builder.Join do
   @moduledoc false
 
   alias Ecto.Query.Builder
-  alias Ecto.Query.{JoinExpr, QueryExpr}
+  alias Ecto.Query.{BooleanExpr, JoinExpr}
 
   @doc """
   Escapes a join expression (not including the `on` expression).
@@ -262,7 +262,7 @@ defmodule Ecto.Query.Builder.Join do
 
       Ecto.Query.Builder.Join.join!(
         query,
-        %JoinExpr{unquote_splicing(join), on: %QueryExpr{}},
+        %JoinExpr{unquote_splicing(join), on: %BooleanExpr{op: :and}},
         unquote(var),
         unquote(as),
         unquote(count_bind),
@@ -286,7 +286,8 @@ defmodule Ecto.Query.Builder.Join do
 
             %JoinExpr{
               unquote_splicing(join),
-              on: %QueryExpr{
+              on: %BooleanExpr{
+                op: :and,
                 expr: unquote(on_expr),
                 params: unquote(on_params),
                 line: unquote(env.line),
@@ -386,7 +387,7 @@ defmodule Ecto.Query.Builder.Join do
 
     join = %{
       join
-      | on: %QueryExpr{expr: on_expr, params: on_params, line: on_line, file: on_file}
+      | on: %BooleanExpr{op: :and, expr: on_expr, params: on_params, line: on_line, file: on_file}
     }
 
     apply(query, join, as, count_bind)
diff --git a/lib/ecto/query/inspect.ex b/lib/ecto/query/inspect.ex
index d5427ff0f0..df7d428c1f 100644
--- a/lib/ecto/query/inspect.ex
+++ b/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.{BooleanExpr, DynamicExpr, JoinExpr, WithExpr, LimitExpr}
 
 defimpl Inspect, for: Ecto.Query.DynamicExpr do
   def inspect(%DynamicExpr{binding: binding} = dynamic, opts) do
@@ -191,8 +191,8 @@ 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)]
+  defp maybe_on(%BooleanExpr{expr: true}, _names), do: []
+  defp maybe_on(%BooleanExpr{} = on, names), do: [on: expr(on, names)]
 
   defp preloads([]), do: []
   defp preloads(preloads), do: [preload: inspect(preloads)]
diff --git a/lib/ecto/query/planner.ex b/lib/ecto/query/planner.ex
index 1615f78d16..685700e695 100644
--- a/lib/ecto/query/planner.ex
+++ b/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}
@@ -49,12 +49,31 @@ defmodule Ecto.Query.Planner do
 
   defp merge_expr_and_params(
          op,
-         %QueryExpr{expr: left_expr, params: left_params} = struct,
-         right_expr,
-         right_params
+         %BooleanExpr{expr: left_expr, params: left_params, subqueries: left_subqueries} = struct,
+         %BooleanExpr{expr: right_expr, params: right_params, subqueries: 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
+      | expr: merge_expr(op, left_expr, right_expr),
+        params: left_params ++ right_params,
+        subqueries: left_subqueries ++ right_subqueries
+    }
+  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 +246,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 +766,8 @@ 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([%{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 +899,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 +985,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)
 
diff --git a/test/ecto/query/planner_test.exs b/test/ecto/query/planner_test.exs
index df91554426..b403ca4818 100644
--- a/test/ecto/query/planner_test.exs
+++ b/test/ecto/query/planner_test.exs
@@ -613,7 +613,7 @@ defmodule Ecto.Query.PlannerTest do
              {:where, [{:and, {:is_nil, [], [nil]}}, {:or, {:is_nil, [], [nil]}}]},
              {:join,
               [
-                {:inner, {"comments", Comment, 38_292_156, "world"}, true, ["join hint"]}
+                {:inner, {"comments", Comment, 38_292_156, "world"}, {:and, true}, ["join hint"]}
               ]},
              {:from, {"posts", Post, 50_009_106, "hello"}, ["hint"]},
              {:select, 1}
@@ -648,6 +648,69 @@ 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
+
+    assert [%{on: %Ecto.Query.BooleanExpr{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()
@@ -971,7 +1034,7 @@ defmodule Ecto.Query.PlannerTest do
 
     assert [
              :all,
-             {:join, [{:inner, {{:fragment, _, _}, Post, _, _}, {:==, _, _}, []}]},
+             {:join, [{:inner, {{:fragment, _, _}, Post, _, _}, {:and, {:==, _, _}}, []}]},
              {:from, {{:fragment, _, _}, Barebone, _, _}, []},
              {:select, {:{}, [], [{:&, [], [0]}, {:&, [], [1]}]}}
            ] = cache_key