Repository: elixir-ecto/ecto
Pull request #2472 (2018-03-22 19:25:37)

Queries in Associations
There wasn't much to do to make this work as expected, because everything aside from the validations was expecting associations to be expressed as queryables anyway (as far as I can tell). I have a few test failures that I'm not 100% sure on, but they appear to be due to changes I made to test schemas. I didn't want to just change them off hand though, because for all I know I broke something real. So I left those tests failing, and wrote new tests to express both the capability of building the query with queries for assocs, and the planning of it. I was going to write integration tests for it, but I figured that if the planner is breaking it out into the correct joins, that it would be unnecessary to have an integration test.

This is in service of providing a lot of the value #2395 was looking for, without having to introduce the larger change of default where clauses.

ebert[bot]: Ebert has finished reviewing this Pull Request and has found:

* 1 possible new issue (including those that may have been commented here).

You can see more details about this review at https://ebertapp.io/github/elixir-ecto/ecto/pulls/2472.

<!-- Ebert auto generated comment -->

zachdaniel: Additionally, I added a validation for only allowing certain fields of the provided query to be modified, but I'm not really sure if thats a good way to do it or not. I also wasn't sure if perhaps we should just *ignore* any additional values in the query, in which case that validation would not be necessary. Due to that, I have not yet written tests for it.
josevalim: Thanks @zachdaniel!

The code looks good but I think we missed the proper coverage with tests. Please take a look at the existing tests in "test/ecto/associations_test.exs". Notice how they are checking for a specific behaviour in assoc_query/join_query. Those are the tests we need to write. For each association type, we need to add a test that covers `assoc_query` and `join_query` using the new association definition. Note there we use `inspect` to compare different queries without the line limits.

At the end of the same file, there are preload tests, that we should add too.

In the planner test, we should add an integration test, that would verify the new mechanism works via a complex join. Imagine something like:

    from a in A, join: b in assoc(a, :b), join: c in assoc(b, :c), join: d in assoc(c, :d)

Where each join uses a queryable association. This is the most important test to have because we need to make sure all of the query bindings are properly rewritten.

In all cases, please avoid adding new schemas to the test files.
zachdaniel: @josevalim I'm curious, it looks like when I provide a query for `assoc_query/3`, like `Ecto.Association.Has.assoc_query(assoc, query, [])`, that the conditions in the association are not honored (which I can see why that would be happening). It also looks like the `Ecto.assoc` helper and friends all pass in explicitly `nil`. With that in mind, I'm not sure if it is correct or incorrect for a query passed into that function to override `where` clauses on the configured query. Thoughts?
josevalim: @zachdaniel yes, we need to "port" the conditions in `where` over to the query. :)
josevalim: We probably need to do something similar to join queries as well, porting the where conditions to an `on` clause.
zachdaniel: Awesome, will do :)
zachdaniel: Is there perhaps some utility for porting where's from one query to another that I should use instead? Along that same line of questioning, I will need to rewrite the where clauses from the join_through to `on` clauses, which I know is something that happens somewhere in ecto ( like when you join on a query), so I'll need to find a way to tap into that.
josevalim: @zachdaniel it is no bother at all!

> I'm trying to determine the context in which additional queries can be passed into assoc_query/3 it looks like that is only done during preload.

Yes, I do think this is the case!

> Seeing as it would make no sense to have the source of a query that isn't the same as the destination of the preload, I think that (in the case of everything but many_to_many, haven't looked at that yet) its as simple as appending the where clauses from the default query onto the provided query.

Beautiful! Agreed.

There is no utility, just concatenate the wheres. :)
zachdaniel: I'm realizing I can use this same logic for the join_through and the existing stuff will adapt, so I think I'll be able to take care of the rest of it.
ebert[bot]: Ebert has finished reviewing this Pull Request and has found:

* 1 possible new issue (including those that may have been commented here).

You can see more details about this review at https://ebertapp.io/github/elixir-ecto/ecto/pulls/2472.

<!-- Ebert auto generated comment -->

zachdaniel: @josevalim so I *think* I've addressed everything you mentioned, and I added a (perhaps pathological) test case in the planner test, and implemented the necessary code to combine where clauses when an override is provided. I still have two failing tests in planner_test and I'm really not sure why. It looks like one of them might have something to do with a cache key that might be based on the fields/associations on the struct, so should I just hardcode the new value? The other one also seems to be testing a complete list of selected fields, so I think I might just need to add the new fields I added to the schema. Any thoughts?
josevalim: @zachdaniel beautiful. I have left two minor comments. Then we just need to write documentation and we are good to go! :heart:
josevalim: @zachdaniel I have noticed that you are relying on the join queries for subqueries. This is actually quite interesting because it means it should be possible to lift the where restriction.

But before we lift any restriction, which I don't think we should now anyway, do you think we will have a difference in performance? Imagine those two queries:

    from s in Summary, join: p in ^(from p in Post, where: p.upvotes >= 1000), on: p.summary_id == s.id

    from s in Summary, join: p in Post, on: p.summary_id == s.id and p.upvotes >= 1000

Which one would be more efficient? Or is it not relevant enough to matter? Could you please investigate by analyzing the generated SQL?
zachdaniel: I'll analyze the SQL tonight, and look at the plan difference. From my experience, I would not expect the performance difference to be significant, because we aren't really using subqueries, we are just using where vs on. If there is a performance difference, I don't think it would be predictably for one way or the other, but more dependent on the specific usage.
zachdaniel: @josevalim So since its not using subqueries, but just using the queries as the subject of the join, it actually generates the same sql! The only difference is that the order of the expressions in the `on` are different. So one says `ON (post.id = summary.post_id) and (p.upvotes > 1000)` and the other one says `ON (p.upvotes > 1000) and (post.id = summary.post_id)`, I doubt that would affect performance in any way.
josevalim: @zachdaniel perfect! This means we can generalize this later to include things like order_by and what not without much hassle. @zachdaniel, we are missing just the docs and we are good to go! A one paragraph description to each association function should be enough. :)
zachdaniel: @josevalim I added some documentation. I'm sure the wording could be improved, so I'll do another pass through later this evening, but wanted to make sure I hit the key points.
zachdaniel: @josevalim pretty much just replaced mine with what you provided :) Should be good to go!

Patch:
diff --git a/lib/ecto/association.ex b/lib/ecto/association.ex
index 09e74373..a20d50e9 100644
--- a/lib/ecto/association.ex
+++ b/lib/ecto/association.ex
@@ -210,6 +210,19 @@ defmodule Ecto.Association do
     |> distinct([x], true)
   end
 
+  @doc """
+  Add the default assoc query where clauses a provided query
+  """
+  def combine_assoc_query(assoc_queryable, nil), do: assoc_queryable
+  def combine_assoc_query(%Ecto.Query{wheres: assoc_wheres}, provided_queryable) do
+    provided_queryable
+    |> Ecto.Queryable.to_query()
+    |> Map.update!(:wheres, &Kernel.++(&1, assoc_wheres))
+  end
+  def combine_assoc_query(_, provided_queryable) do
+    provided_queryable
+  end
+
   defp assoc_to_where(%{on: %QueryExpr{} = on}) do
     on
     |> Map.put(:__struct__, BooleanExpr)
@@ -249,21 +262,31 @@ defmodule Ecto.Association do
 
   ## Examples
 
-      iex> Ecto.Association.related_from_query({"custom_source", Schema})
+      iex> Ecto.Association.related_from_query({"custom_source", Schema}, :comments_v1)
       Schema
 
-      iex> Ecto.Association.related_from_query(Schema)
+      iex> Ecto.Association.related_from_query(Schema, :comments_v1)
       Schema
 
-      iex> Ecto.Association.related_from_query("wrong")
-      ** (ArgumentError) association queryable must be a schema or {source, schema}, got: "wrong"
-
+      iex> Ecto.Association.related_from_query("wrong", :comments_v1)
+      ** (ArgumentError) association :comments_v1 queryable must be a schema, a {source, schema}, or a query. got: "wrong"
   """
-  def related_from_query(atom) when is_atom(atom), do: atom
-  def related_from_query({source, schema}) when is_binary(source) and is_atom(schema), do: schema
-  def related_from_query(queryable) do
-    raise ArgumentError, "association queryable must be a schema " <>
-      "or {source, schema}, got: #{inspect queryable}"
+  def related_from_query(atom, _name) when is_atom(atom), do: atom
+  def related_from_query({source, schema}, _name) when is_binary(source) and is_atom(schema), do: schema
+  def related_from_query(%Ecto.Query{from: {source, schema}} = query, name) when is_binary(source) and is_atom(schema) do
+    case query do
+      %Ecto.Query{order_bys: [], limit: nil, offset: nil, group_bys: [], joins: [],
+                  havings: [], preloads: [], assocs: [], distinct: nil, lock: nil} ->
+        schema
+      _ ->
+        raise ArgumentError,
+          "A query was provided for many_to_many #{inspect name}, but that query included a statement other" <>
+          "than a `where` clause. Queries in :join_through only support `where` clauses, nothing else."
+    end
+  end
+  def related_from_query(queryable, name) do
+    raise ArgumentError, "association #{inspect name} queryable must be a schema, " <>
+      "a {source, schema}, or a query. got: #{inspect queryable}"
   end
 
   @doc """
@@ -460,7 +483,7 @@ defmodule Ecto.Association.Has do
 
     queryable = Keyword.fetch!(opts, :queryable)
     cardinality = Keyword.fetch!(opts, :cardinality)
-    related = Ecto.Association.related_from_query(queryable)
+    related = Ecto.Association.related_from_query(queryable, name)
 
     if opts[:through] do
       raise ArgumentError, "invalid association #{inspect name}. When using the :through " <>
@@ -519,13 +542,14 @@ defmodule Ecto.Association.Has do
 
   @doc false
   def assoc_query(%{queryable: queryable, related_key: related_key}, query, [value]) do
-    from x in (query || queryable),
+
+    from x in Ecto.Association.combine_assoc_query(queryable, query),
       where: field(x, ^related_key) == ^value
   end
 
   @doc false
   def assoc_query(%{queryable: queryable, related_key: related_key}, query, values) do
-    from x in (query || queryable),
+    from x in Ecto.Association.combine_assoc_query(queryable, query),
       where: field(x, ^related_key) in ^values
   end
 
@@ -737,7 +761,7 @@ defmodule Ecto.Association.BelongsTo do
   def struct(module, name, opts) do
     ref       = if ref = opts[:references], do: ref, else: :id
     queryable = Keyword.fetch!(opts, :queryable)
-    related   = Ecto.Association.related_from_query(queryable)
+    related   = Ecto.Association.related_from_query(queryable, name)
 
     unless is_atom(related) do
       raise ArgumentError, "association queryable must be a schema, got: #{inspect related}"
@@ -780,13 +804,13 @@ defmodule Ecto.Association.BelongsTo do
 
   @doc false
   def assoc_query(%{queryable: queryable, related_key: related_key}, query, [value]) do
-    from x in (query || queryable),
+    from x in Ecto.Association.combine_assoc_query(queryable, query),
       where: field(x, ^related_key) == ^value
   end
 
   @doc false
   def assoc_query(%{queryable: queryable, related_key: related_key}, query, values) do
-    from x in (query || queryable),
+    from x in Ecto.Association.combine_assoc_query(queryable, query),
       where: field(x, ^related_key) in ^values
   end
 
@@ -884,17 +908,11 @@ defmodule Ecto.Association.ManyToMany do
   def struct(module, name, opts) do
     join_through = opts[:join_through]
 
-    if join_through && (is_atom(join_through) or is_binary(join_through)) do
-      :ok
-    else
-      raise ArgumentError,
-        "many_to_many #{inspect name} associations require the :join_through option to be " <>
-        "given and it must be an atom (representing a schema) or a string (representing a table)"
-    end
+    validate_join_through(name, join_through)
 
     join_keys = opts[:join_keys]
     queryable = Keyword.fetch!(opts, :queryable)
-    related   = Ecto.Association.related_from_query(queryable)
+    related   = Ecto.Association.related_from_query(queryable, name)
 
     {owner_key, join_keys} =
       case join_keys do
@@ -975,7 +993,7 @@ defmodule Ecto.Association.ManyToMany do
     # We need to go all the way using owner and query so
     # Ecto has all the information necessary to cast fields.
     # This also helps validate the associated schema exists all the way.
-    from q in (query || queryable),
+    from q in Ecto.Association.combine_assoc_query(queryable, query),
       join: o in ^owner, on: field(o, ^owner_key) in ^values,
       join: j in ^join_through, on: field(j, ^join_owner_key) == field(o, ^owner_key),
       where: field(j, ^join_related_key) == field(q, ^related_key)
@@ -1049,6 +1067,30 @@ defmodule Ecto.Association.ManyToMany do
     end
   end
 
+  defp validate_join_through(name, nil) do
+    raise ArgumentError, "many_to_many #{inspect name} associations require the :join_through option to be given"
+  end
+  defp validate_join_through(name, %Ecto.Query{} = query) do
+    case query do
+      %Ecto.Query{order_bys: [], limit: nil, offset: nil, group_bys: [], joins: [],
+                  havings: [], preloads: [], assocs: [], distinct: nil, lock: nil} ->
+        query
+      _ ->
+        raise ArgumentError,
+          "A query was provided for many_to_many #{inspect name}, but that query included a statement other" <>
+          "than a `where` clause. Queries in :join_through only support `where` clauses, nothing else."
+    end
+  end
+  defp validate_join_through(_, join_through) when is_atom(join_through) or is_binary(join_through) do
+    :ok
+  end
+  defp validate_join_through(name, _join_through) do
+    raise ArgumentError,
+      "many_to_many #{inspect name} associations require the :join_through option to be " <>
+      "an atom (representing a schema), a string (representing a table) or a query" <>
+      "with only `where` clauses."
+  end
+
   defp insert_join?(%{action: :insert}, _, _field, _related_key), do: true
   defp insert_join?(_, %{action: :insert}, _field, _related_key), do: true
   defp insert_join?(%{data: owner}, %{data: related}, field, related_key) do
diff --git a/lib/ecto/schema.ex b/lib/ecto/schema.ex
index de7d6d9c..d15a87d6 100644
--- a/lib/ecto/schema.ex
+++ b/lib/ecto/schema.ex
@@ -650,6 +650,30 @@ defmodule Ecto.Schema do
         end
       end
 
+  ## Using Queries as Associations
+
+  A query can also be given instead of a schema. Querying, joining or preloading the association will
+  use the given query. Currently only where clauses can be provided in queries. Let's see an example:
+
+      defmodule Comment do
+        ...
+        def deleted() do
+          from comment in Comment,
+            where: not(is_nil(comment.deleted_at))
+        end
+      end
+
+      defmodule Post do
+        use Ecto.Schema
+        schema "posts" do
+          has_many :deleted_comments, Comment.deleted()
+        end
+      end
+
+  Note: building the association does not consider the query filters.
+  For example, if the given query requires the active field of the associated records to be true,
+  building such association won't automatically set the active field to true.
+
   ## has_many/has_one :through
 
   Ecto also supports defining associations in terms of other associations
@@ -791,6 +815,29 @@ defmodule Ecto.Schema do
       [post] = Repo.all(from(p in Post, where: p.id == 42, preload: :permalink))
       post.permalink #=> %Permalink{...}
 
+  ## Using Queries as Associations
+
+  A query can also be given instead of a schema. Querying, joining or preloading the association will
+  use the given query. Currently only where clauses can be provided in queries. Let's see an example:
+
+      defmodule Post do
+        ...
+        def active() do
+          from post in Post,
+            where: post.active
+        end
+      end
+
+      defmodule Comment do
+        use Ecto.Schema
+        schema "comments" do
+          has_one :post, Post.active()
+        end
+      end
+
+  Note: building the association does not consider the query filters.
+  For example, if the given query requires the active field of the associated records to be true,
+  building such association won't automatically set the active field to true.
   """
   defmacro has_one(name, queryable, opts \\ []) do
     queryable = expand_alias(queryable, __CALLER__)
@@ -864,6 +911,30 @@ defmodule Ecto.Schema do
         end
       end
 
+  ## Using Queries as Associations
+
+  A query can also be given instead of a schema. Querying, joining or preloading the association will
+  use the given query. Currently only where clauses can be provided in queries. Let's see an example:
+
+      defmodule Post do
+        ...
+        def active() do
+          from post in Post,
+            where: post.active
+        end
+      end
+
+      defmodule Comment do
+        use Ecto.Schema
+        schema "posts" do
+          belongs_to :post, Post.active()
+        end
+      end
+
+  Note: building the association does not consider the query filters.
+  For example, if the given query requires the active field of the associated records to be true,
+  building such association won't automatically set the active field to true.
+
   ## Polymorphic associations
 
   One common use case for belongs to associations is to handle
@@ -1161,6 +1232,56 @@ defmodule Ecto.Schema do
         {:ok, assoc} -> # Assoc was created!
         {:error, changeset} -> # Handle the error
       end
+
+  ## Using Queries as Associations
+
+  A query can also be given instead of a schema, both for the join_through and the destination.
+  Querying, joining or preloading the association will use the given query. Currently only where
+  clauses can be provided in queries. Let's see an example:
+
+      defmodule UserOrganization do
+        use Ecto.Schema
+
+        @primary_key false
+        schema "users_organizations" do
+          belongs_to :user, User
+          belongs_to :organization, Organization
+
+          field :deleted, :boolean
+          timestamps # Added bonus, a join schema will also allow you to set timestamps
+        end
+
+        def active() do
+          from user_organization in UserOrganization,
+            where: is_nil(user_organization.deleted)
+        end
+      end
+
+      defmodule User do
+        use Ecto.Schema
+
+        schema "users" do
+          many_to_many :organizations, Organization, join_through: UserOrganization
+          field :banned, :boolean
+        end
+
+        def not_banned() do
+          from user in User,
+            where: not(user.banned)
+        end
+      end
+
+      defmodule Organization do
+        use Ecto.Schema
+
+        schema "organizations" do
+          many_to_many :users, User.not_banned(), join_through: UserOrganization.active()
+        end
+      end
+
+  Note: building the association does not consider the query filters.
+  For example, if the given query requires the active field of the associated records to be true,
+  building such association won't automatically set the active field to true.
   """
   defmacro many_to_many(name, queryable, opts \\ []) do
     quote do


Test patch:
diff --git a/test/ecto/association_test.exs b/test/ecto/association_test.exs
index eb90107f..50afdf90 100644
--- a/test/ecto/association_test.exs
+++ b/test/ecto/association_test.exs
@@ -21,6 +21,7 @@ defmodule Ecto.AssociationTest do
 
     schema "posts" do
       field :title, :string
+      field :upvotes, :integer
 
       has_many :comments, Comment
       has_one :permalink, Permalink
@@ -28,6 +29,11 @@ defmodule Ecto.AssociationTest do
       belongs_to :author, Author, defaults: [title: "World!"]
       belongs_to :summary, Summary
     end
+
+    def awesome() do
+      from row in __MODULE__,
+        where: row.upvotes >= 1000
+    end
   end
 
   defmodule Comment do
@@ -48,9 +54,15 @@ defmodule Ecto.AssociationTest do
 
     schema "permalinks" do
       field :url, :string
+      field :special, :boolean
       many_to_many :authors, Author, join_through: AuthorPermalink, defaults: [title: "m2m!"]
       has_many :author_emails, through: [:authors, :emails]
     end
+
+    def special() do
+      from row in __MODULE__,
+        where: row.special
+    end
   end
 
   defmodule PostWithPrefix do
@@ -72,30 +84,50 @@ defmodule Ecto.AssociationTest do
     end
   end
 
+  defmodule AuthorPermalink do
+    use Ecto.Schema
+
+    schema "authors_permalinks" do
+      field :author_id
+      field :permalink_id
+      field :deleted, :boolean
+    end
+
+    def active() do
+      from row in __MODULE__,
+        where: not(row.deleted)
+    end
+  end
+
   defmodule Author do
     use Ecto.Schema
 
     schema "authors" do
       field :title, :string
+      field :super_user, :boolean
       has_many :posts, Post, on_replace: :delete
       has_many :posts_comments, through: [:posts, :comments]    # many -> many
       has_many :posts_permalinks, through: [:posts, :permalink] # many -> one
       has_many :emails, {"users_emails", Email}
+      has_many :awesome_posts, Post.awesome()
       has_one :profile, {"users_profiles", Profile},
         defaults: [name: "default"], on_replace: :delete
       many_to_many :permalinks, {"custom_permalinks", Permalink},
         join_through: "authors_permalinks"
+
+      many_to_many :active_special_permalinks, Permalink.special(),
+        join_through: AuthorPermalink.active()
+      many_to_many :special_permalinks, Permalink.special(),
+        join_through: AuthorPermalink
+      many_to_many :active_permalinks, Permalink,
+        join_through: AuthorPermalink.active()
       has_many :posts_with_prefix, PostWithPrefix
       has_many :comments_with_prefix, through: [:posts_with_prefix, :comments_with_prefix]
     end
-  end
-
-  defmodule AuthorPermalink do
-    use Ecto.Schema
 
-    schema "authors_permalinks" do
-      field :author_id
-      field :permalink_id
+    def super_users() do
+      from row in __MODULE__,
+        where: row.super_user
     end
   end
 
@@ -104,6 +136,7 @@ defmodule Ecto.AssociationTest do
 
     schema "summaries" do
       has_one :post, Post, defaults: [title: "default"], on_replace: :nilify
+      has_one :awesome_post, Post.awesome()
       has_many :posts, Post, on_replace: :nilify
       has_one :post_author, through: [:post, :author]        # one -> belongs
       has_many :post_comments, through: [:post, :comments]   # one -> many
@@ -115,6 +148,7 @@ defmodule Ecto.AssociationTest do
 
     schema "emails" do
       belongs_to :author, {"post_authors", Author}
+      belongs_to :super_user, Author.super_users(), foreign_key: :author_id, define_field: false
     end
   end
 
@@ -161,6 +195,25 @@ defmodule Ecto.AssociationTest do
            inspect(from c in Comment, where: c.post_id in ^[1, 2, 3], limit: 5)
   end
 
+  test "has many query-based assoc query" do
+    assoc = Author.__schema__(:association, :awesome_posts)
+    assert inspect(Ecto.Association.Has.joins_query(assoc)) ==
+           inspect(from a in Author, join: p in ^(from p in Post, where: p.upvotes >= 1000), on: p.author_id == a.id)
+
+    assert inspect(Ecto.Association.Has.assoc_query(assoc, nil, [])) ==
+           inspect(from p in Post, where: p.upvotes >= 1000, where: p.author_id in ^[])
+
+    assert inspect(Ecto.Association.Has.assoc_query(assoc, nil, [1, 2, 3])) ==
+           inspect(from p in Post, where: p.upvotes >= 1000, where: p.author_id in ^[1, 2, 3])
+  end
+
+  test "has many query-based assoc with custom query" do
+    assoc = Author.__schema__(:association, :awesome_posts)
+    query = from p in Post, limit: 5
+    assert inspect(Ecto.Association.Has.assoc_query(assoc, query, [1, 2, 3])) ==
+           inspect(from p in Post, where: p.upvotes >= 1000, where: p.author_id in ^[1, 2, 3], limit: 5)
+  end
+
   test "has one" do
     assoc = Post.__schema__(:association, :permalink)
 
@@ -197,6 +250,25 @@ defmodule Ecto.AssociationTest do
            inspect(from c in Permalink, where: c.post_id in ^[1, 2, 3], limit: 5)
   end
 
+  test "has one query-based assoc query" do
+    assoc = Summary.__schema__(:association, :awesome_post)
+    assert inspect(Ecto.Association.Has.joins_query(assoc)) ==
+           inspect(from s in Summary, join: p in ^(from p in Post, where: p.upvotes >= 1000), on: p.summary_id == s.id)
+
+    assert inspect(Ecto.Association.Has.assoc_query(assoc, nil, [])) ==
+           inspect(from p in Post, where: p.upvotes >= 1000, where: p.summary_id in ^[])
+
+    assert inspect(Ecto.Association.Has.assoc_query(assoc, nil, [1, 2, 3])) ==
+           inspect(from p in Post, where: p.upvotes >= 1000, where: p.summary_id in ^[1, 2, 3])
+  end
+
+  test "has one query-based assoc with custom query" do
+    assoc = Summary.__schema__(:association, :awesome_post)
+    query = from p in Post, limit: 5
+    assert inspect(Ecto.Association.Has.assoc_query(assoc, query, [1, 2, 3])) ==
+           inspect(from p in Post, where: p.upvotes >= 1000, where: p.summary_id in ^[1, 2, 3], limit: 5)
+  end
+
   test "belongs to" do
     assoc = Post.__schema__(:association, :author)
 
@@ -236,6 +308,25 @@ defmodule Ecto.AssociationTest do
            inspect(from a in Author, where: a.id in ^[1, 2, 3], limit: 5)
   end
 
+  test "belongs to query-based assoc query" do
+    assoc = Email.__schema__(:association, :super_user)
+    assert inspect(Ecto.Association.BelongsTo.joins_query(assoc)) ==
+           inspect(from e in Email, join: a in ^(from a in Author, where: a.super_user), on: a.id == e.author_id)
+
+    assert inspect(Ecto.Association.BelongsTo.assoc_query(assoc, nil, [])) ==
+           inspect(from a in Author, where: a.super_user, where: a.id in ^[])
+
+    assert inspect(Ecto.Association.BelongsTo.assoc_query(assoc, nil, [1, 2, 3])) ==
+           inspect(from a in Author, where: a.super_user, where: a.id in ^[1, 2, 3])
+  end
+
+  test "belongs to query-based assoc with custom query" do
+    assoc = Email.__schema__(:association, :super_user)
+    query = from a in Author, limit: 5
+    assert inspect(Ecto.Association.BelongsTo.assoc_query(assoc, query, [1, 2, 3])) ==
+           inspect(from a in Author, where: a.super_user, where: a.id in ^[1, 2, 3], limit: 5)
+  end
+
   test "many to many" do
     assoc = Permalink.__schema__(:association, :authors)
 
@@ -288,6 +379,125 @@ defmodule Ecto.AssociationTest do
                     where: m.author_id == a.id, limit: 5)
   end
 
+  test "many to many query-based assoc and query based join_through query" do
+    assoc = Author.__schema__(:association, :active_special_permalinks)
+    assert inspect(Ecto.Association.ManyToMany.joins_query(assoc)) ==
+           inspect(from a in Author,
+                    join: m in ^(from m in AuthorPermalink, where: not(m.deleted)),
+                    on: m.author_id == a.id,
+                    join: p in ^(from p in Permalink, where: p.special),
+                    on: m.permalink_id == p.id)
+
+    assert inspect(Ecto.Association.ManyToMany.assoc_query(assoc, nil, [])) ==
+           inspect(from p in Permalink, where: p.special,
+                    join: a in Author, on: a.id in ^[],
+                    join: m in ^(from m in AuthorPermalink, where: not(m.deleted)),
+                    on: m.author_id == a.id,
+                    where: m.permalink_id == p.id
+             )
+
+    assert inspect(Ecto.Association.ManyToMany.assoc_query(assoc, nil, [1, 2, 3])) ==
+           inspect(from p in Permalink, where: p.special,
+                    join: a in Author, on: a.id in ^[1, 2, 3],
+                    join: m in ^(from m in AuthorPermalink, where: not(m.deleted)),
+                    on: m.author_id == a.id,
+                    where: m.permalink_id == p.id
+             )
+  end
+
+  test "many to many query-based assoc and query based join through with custom query" do
+    assoc = Author.__schema__(:association, :active_special_permalinks)
+    query = from p in Permalink, limit: 5
+    assert inspect(Ecto.Association.ManyToMany.assoc_query(assoc, query, [1, 2, 3])) ==
+           inspect(from p in Permalink,
+                    join: a in Author,
+                    on: a.id in ^[1, 2, 3],
+                    join: m in ^(from m in AuthorPermalink, where: not(m.deleted)),
+                    on: m.author_id == a.id,
+                    where: p.special(),
+                    where: m.permalink_id == p.id,
+                    limit: 5)
+  end
+
+  test "many to many query-based assoc query" do
+    assoc = Author.__schema__(:association, :special_permalinks)
+    assert inspect(Ecto.Association.ManyToMany.joins_query(assoc)) ==
+           inspect(from a in Author,
+                    join: m in AuthorPermalink,
+                    on: m.author_id == a.id,
+                    join: p in ^(from p in Permalink, where: p.special),
+                    on: m.permalink_id == p.id)
+
+    assert inspect(Ecto.Association.ManyToMany.assoc_query(assoc, nil, [])) ==
+           inspect(from p in Permalink, where: p.special,
+                    join: a in Author, on: a.id in ^[],
+                    join: m in AuthorPermalink,
+                    on: m.author_id == a.id,
+                    where: m.permalink_id == p.id
+             )
+
+    assert inspect(Ecto.Association.ManyToMany.assoc_query(assoc, nil, [1, 2, 3])) ==
+           inspect(from p in Permalink, where: p.special,
+                    join: a in Author, on: a.id in ^[1, 2, 3],
+                    join: m in AuthorPermalink,
+                    on: m.author_id == a.id,
+                    where: m.permalink_id == p.id
+             )
+  end
+
+  test "many to many query-based assoc query with custom query" do
+    assoc = Author.__schema__(:association, :special_permalinks)
+    query = from p in Permalink, limit: 5
+    assert inspect(Ecto.Association.ManyToMany.assoc_query(assoc, query, [1, 2, 3])) ==
+           inspect(from p in Permalink,
+                    join: a in Author,
+                    on: a.id in ^[1, 2, 3],
+                    join: m in AuthorPermalink,
+                    on: m.author_id == a.id,
+                    where: p.special(),
+                    where: m.permalink_id == p.id,
+                    limit: 5)
+  end
+
+  test "many to many query-based join through query" do
+    assoc = Author.__schema__(:association, :active_permalinks)
+    assert inspect(Ecto.Association.ManyToMany.joins_query(assoc)) ==
+           inspect(from a in Author,
+                    join: m in ^(from m in AuthorPermalink, where: not(m.deleted)),
+                    on: m.author_id == a.id,
+                    join: p in Permalink,
+                    on: m.permalink_id == p.id)
+
+    assert inspect(Ecto.Association.ManyToMany.assoc_query(assoc, nil, [])) ==
+           inspect(from p in Permalink,
+                    join: a in Author, on: a.id in ^[],
+                    join: m in ^(from m in AuthorPermalink, where: not(m.deleted)),
+                    on: m.author_id == a.id,
+                    where: m.permalink_id == p.id
+             )
+
+    assert inspect(Ecto.Association.ManyToMany.assoc_query(assoc, nil, [1, 2, 3])) ==
+           inspect(from p in Permalink,
+                    join: a in Author, on: a.id in ^[1, 2, 3],
+                    join: m in ^(from m in AuthorPermalink, where: not(m.deleted)),
+                    on: m.author_id == a.id,
+                    where: m.permalink_id == p.id
+             )
+  end
+
+  test "many to many query-based join through with custom query" do
+    assoc = Author.__schema__(:association, :active_permalinks)
+    query = from p in Permalink, limit: 5
+    assert inspect(Ecto.Association.ManyToMany.assoc_query(assoc, query, [1, 2, 3])) ==
+           inspect(from p in Permalink,
+                    join: a in Author,
+                    on: a.id in ^[1, 2, 3],
+                    join: m in ^(from m in AuthorPermalink, where: not(m.deleted)),
+                    on: m.author_id == a.id,
+                    where: m.permalink_id == p.id,
+                    limit: 5)
+  end
+
   test "has many through many to many" do
     assoc = Author.__schema__(:association, :posts_comments)
 
diff --git a/test/ecto/query/planner_test.exs b/test/ecto/query/planner_test.exs
index 4a3a2db6..2e6a3d84 100644
--- a/test/ecto/query/planner_test.exs
+++ b/test/ecto/query/planner_test.exs
@@ -16,8 +16,32 @@ defmodule Ecto.Query.PlannerTest do
       field :temp, :string, virtual: true
       field :posted, :naive_datetime
       field :uuid, :binary_id
+      field :special, :boolean
       belongs_to :post, Ecto.Query.PlannerTest.Post
       has_many :post_comments, through: [:post, :comments]
+      has_many :comment_posts, Ecto.Query.PlannerTest.CommentPost
+    end
+
+    def special() do
+      from comment in __MODULE__,
+        where: comment.special
+    end
+  end
+
+  defmodule CommentPost do
+    use Ecto.Schema
+
+    schema "comment_posts" do
+      belongs_to :comment, Comment
+      belongs_to :post, Post
+      belongs_to :special_comment, Comment.special()
+
+      field :deleted, :boolean
+    end
+
+    def inactive() do
+      from comment_post in __MODULE__,
+        where: comment_post.deleted
     end
   end
 
@@ -34,6 +58,9 @@ defmodule Ecto.Query.PlannerTest do
       field :links, {:array, Custom.Permalink}
       has_many :comments, Ecto.Query.PlannerTest.Comment
       has_many :extra_comments, Ecto.Query.PlannerTest.Comment
+      has_many :special_comments, Ecto.Query.PlannerTest.Comment.special()
+
+      many_to_many :shared_special_comments, Comment.special(), join_through: CommentPost.inactive()
     end
   end
 
@@ -229,6 +256,44 @@ defmodule Ecto.Query.PlannerTest do
     assert Macro.to_string(join3.on.expr) == "&2.id() == &0.post_id()"
   end
 
+  test "prepare: joins associations with custom queries" do
+    query = from(p in Post, left_join: assoc(p, :special_comments)) |> prepare |> elem(0)
+
+    assert {{"posts", _}, {"comments", _}} = query.sources
+    assert [join] = query.joins
+    assert join.ix == 1
+    assert Macro.to_string(join.on.expr) == "&1.special() and &1.post_id() == &0.id()"
+
+    query = from(p in Post, left_join: assoc(p, :shared_special_comments)) |> prepare |> elem(0)
+
+    assert {{"posts", _}, {"comments", _}, {"comment_posts", _}} = query.sources
+    assert [join1, join2] = query.joins
+    assert Enum.map(query.joins, & &1.ix) == [2, 1]
+    assert Macro.to_string(join1.on.expr) == "&2.deleted() and &2.post_id() == &0.id()"
+    assert Macro.to_string(join2.on.expr) == "&1.special() and &2.comment_id() == &1.id()"
+  end
+
+  test "prepare: nested joins associations with custom queries" do
+    query = from(p in Post,
+                   join: c in assoc(p, :special_comments),
+                   join: p2 in assoc(c, :post),
+                   join: c1 in assoc(p, :shared_special_comments),
+                   join: cp in assoc(c1, :comment_posts),
+                   join: c2 in assoc(cp, :special_comment))
+                   |> prepare
+                   |> elem(0)
+
+    assert [join1, join2, join3, join4, join5, join6] = query.joins
+    assert {{"posts", _}, {"comments", _}, {"posts", _}, {"comments", _}, {"comment_posts", _}, {"comments", _}, {"comment_posts", _}} = query.sources
+
+    assert Macro.to_string(join1.on.expr) == "&1.special() and &1.post_id() == &0.id()"
+    assert Macro.to_string(join2.on.expr) == "&2.id() == &1.post_id()"
+    assert Macro.to_string(join3.on.expr) == "&6.deleted() and &6.post_id() == &0.id()"
+    assert Macro.to_string(join4.on.expr) == "&3.special() and &6.comment_id() == &3.id()"
+    assert Macro.to_string(join5.on.expr) == "&4.comment_id() == &3.id()"
+    assert Macro.to_string(join6.on.expr) == "&5.special() and &5.id() == &4.special_comment_id()"
+  end
+
   test "prepare: cannot associate without schema" do
     query   = from(p in "posts", join: assoc(p, :comments))
     message = ~r"cannot perform association join on \"posts\" because it does not have a schema"
@@ -257,7 +322,7 @@ defmodule Ecto.Query.PlannerTest do
                    {:lock, "foo"},
                    {:prefix, "foo"},
                    {:where, [{:and, {:is_nil, [], [nil]}}, {:or, {:is_nil, [], [nil]}}]},
-                   {:join, [{:inner, {"comments", Comment, 6996781}, true}]},
+                   {:join, [{:inner, {"comments", Comment, 48293978}, true}]},
                    {"posts", Post, 27727487},
                    {:select, 1}]
   end
@@ -387,7 +452,7 @@ defmodule Ecto.Query.PlannerTest do
       |> normalize()
     assert query.select.fields ==
            select_fields([:id, :post_title, :text, :code, :posted, :visits, :links], 0) ++
-           select_fields([:id, :text, :posted, :uuid, :post_id], 1) ++
+           select_fields([:id, :text, :posted, :uuid, :special, :post_id], 1) ++
            [{{:., [], [{:&, [], [0]}, :post_title]}, [], []}]
   end
 
diff --git a/test/ecto/schema_test.exs b/test/ecto/schema_test.exs
index cce27759..5b1b2540 100644
--- a/test/ecto/schema_test.exs
+++ b/test/ecto/schema_test.exs
@@ -580,7 +580,7 @@ defmodule Ecto.SchemaTest do
   end
 
   test "has_* expects a queryable" do
-    message = ~r"association queryable must be a schema or {source, schema}, got: 123"
+    message = ~r"association :posts queryable must be a schema, a {source, schema}, or a query. got: 123"
     assert_raise ArgumentError, message, fn ->
       defmodule QueryableMisMatch do
         use Ecto.Schema