diff --git a/Earthfile b/Earthfile
index 630a86cb97..fbdf73792b 100644
--- a/Earthfile
+++ b/Earthfile
@@ -60,7 +60,7 @@ integration-test-base:
         apk del .build-dependencies && rm -f msodbcsql*.sig mssql-tools*.apk
     ENV PATH="/opt/mssql-tools/bin:${PATH}"
 
-    GIT CLONE https://github.com/elixir-ecto/ecto_sql.git /src/ecto_sql
+    GIT CLONE --branch composite_foreign_keys https://github.com/soundmonster/ecto_sql.git /src/ecto_sql
     WORKDIR /src/ecto_sql
     RUN mix deps.get
 
diff --git a/integration_test/cases/assoc.exs b/integration_test/cases/assoc.exs
index fe2eeb1927..f002d8c351 100644
--- a/integration_test/cases/assoc.exs
+++ b/integration_test/cases/assoc.exs
@@ -10,6 +10,8 @@ defmodule Ecto.Integration.AssocTest do
   alias Ecto.Integration.PostUser
   alias Ecto.Integration.Comment
   alias Ecto.Integration.Permalink
+  alias Ecto.Integration.CompositePk
+  alias Ecto.Integration.OneToOneCompositePk
 
   test "has_many assoc" do
     p1 = TestRepo.insert!(%Post{title: "1"})
@@ -42,6 +44,21 @@ defmodule Ecto.Integration.AssocTest do
     assert l3.id == lid3
   end
 
+  test "has_one assoc with composite key" do
+    c11 = TestRepo.insert!(%CompositePk{a: 1, b: 1, name: "11"})
+    _c12 = TestRepo.insert!(%CompositePk{a: 1, b: 2, name: "12"})
+    c22 = TestRepo.insert!(%CompositePk{a: 2, b: 2, name: "22"})
+
+    %OneToOneCompositePk{id: id_o11} = TestRepo.insert!(%OneToOneCompositePk{composite_a: 1, composite_b: 1})
+    %OneToOneCompositePk{} = TestRepo.insert!(%OneToOneCompositePk{composite_a: 1, composite_b: 2})
+    %OneToOneCompositePk{id: id_o22} = TestRepo.insert!(%OneToOneCompositePk{composite_a: 2, composite_b: 2})
+
+    [o11, o22] = TestRepo.all(Ecto.assoc([c11, c22], :one_to_one_composite_pk))
+    assert o11.id == id_o11
+    assert o22.id == id_o22
+  end
+
+
   test "belongs_to assoc" do
     %Post{id: pid1} = TestRepo.insert!(%Post{title: "1"})
     %Post{id: pid2} = TestRepo.insert!(%Post{title: "2"})
@@ -55,6 +72,22 @@ defmodule Ecto.Integration.AssocTest do
     assert p2.id == pid2
   end
 
+  test "belongs_to assoc with composite key" do
+    TestRepo.insert!(%CompositePk{a: 2, b: 1, name: "foo"})
+    TestRepo.insert!(%CompositePk{a: 2, b: 2, name: "bar"})
+    TestRepo.insert!(%CompositePk{a: 2, b: 3, name: "unused"})
+
+    p1 = TestRepo.insert!(%Post{title: "first", composite_a: 2, composite_b: 1})
+    p2 = TestRepo.insert!(%Post{title: "none"})
+    p3 = TestRepo.insert!(%Post{title: "second", composite_a: 2, composite_b: 2})
+
+    assert [c1, c2] = TestRepo.all Ecto.assoc([p1, p2, p3], :composite)
+    assert c1.a == 2
+    assert c1.b == 1
+    assert c2.a == 2
+    assert c2.b == 2
+  end
+
   test "has_many through assoc" do
     p1 = TestRepo.insert!(%Post{})
     p2 = TestRepo.insert!(%Post{})
@@ -231,6 +264,22 @@ defmodule Ecto.Integration.AssocTest do
     assert u2.id == uid2
   end
 
+  test "many_to_many composite PK" do
+    c11 = TestRepo.insert!(%CompositePk{a: 1, b: 1, name: "11"})
+    c12 = TestRepo.insert!(%CompositePk{a: 1, b: 2, name: "12"})
+    c22 = TestRepo.insert!(%CompositePk{a: 2, b: 2, name: "22"})
+
+    TestRepo.insert_all "composite_pk_composite_pk", [[a_1: 1, b_1: 1, a_2: 1, b_2: 2],
+                                                      [a_1: 1, b_1: 1, a_2: 2, b_2: 2],
+                                                      [a_1: 1, b_1: 2, a_2: 2, b_2: 2]]
+
+    assert [^c12, ^c22] = TestRepo.all Ecto.assoc([c11], :composites)
+    assert [^c22] = TestRepo.all Ecto.assoc([c12], :composites)
+    assert [] = TestRepo.all Ecto.assoc([c22], :composites)
+
+    assert [^c12, ^c22, ^c22] = TestRepo.all Ecto.assoc([c11, c12, c22], :composites)
+  end
+
   ## Changesets
 
   test "has_one changeset assoc (on_replace: :delete)" do
@@ -725,6 +774,27 @@ defmodule Ecto.Integration.AssocTest do
     assert perma.post_id == nil
   end
 
+  test "belongs_to changeset assoc on composite key" do
+    changeset =
+      %CompositePk{a: 1, b: 2}
+      |> Ecto.Changeset.change()
+      |> Ecto.Changeset.put_assoc(:posts, [%Post{title: "1"}])
+
+    composite = TestRepo.insert!(changeset)
+    assert [post] = composite.posts
+    assert post.id
+    assert post.composite_a == composite.a
+    assert post.composite_b == composite.b
+    assert post.title == "1"
+
+    composite = TestRepo.get_by! from(CompositePk, preload: [:posts]), [a: composite.a, b: composite.b]
+    assert [%Post{title: "1"}] = composite.posts
+
+    post = TestRepo.get! from(Post, preload: [:composite]), post.id
+    assert post.composite.a == 1
+    assert post.composite.b == 2
+  end
+
   test "inserting struct with associations" do
     tree = %Permalink{
       url: "root",
@@ -750,6 +820,14 @@ defmodule Ecto.Integration.AssocTest do
     assert Enum.all?(tree.post.comments, & &1.id)
   end
 
+  test "inserting struct with associations on composite keys" do
+    # creates nested belongs_to
+    %Post{composite: composite} =
+      TestRepo.insert! %Post{title: "1", composite: %CompositePk{a: 1, b: 2, name: "name"}}
+
+    assert %CompositePk{a: 1, b: 2, name: "name"} = composite
+  end
+
   test "inserting struct with empty associations" do
     permalink = TestRepo.insert!(%Permalink{url: "root", post: nil})
     assert permalink.post == nil
diff --git a/integration_test/cases/joins.exs b/integration_test/cases/joins.exs
index 543cf810a3..ee830d7ad7 100644
--- a/integration_test/cases/joins.exs
+++ b/integration_test/cases/joins.exs
@@ -573,7 +573,8 @@ defmodule Ecto.Integration.JoinsTest do
     query = from(p in Post, join: pl in assoc(p, :permalink),
                             join: c in assoc(p, :comments),
                             preload: [permalink: pl],
-                            select: {p, c})
+                            select: {p, c},
+                            order_by: p.id)
     [{p1, ^c1}, {p1, ^c2}, {p2, ^c3}] = TestRepo.all(query)
     assert p1.permalink == pl1
     assert p2.permalink == pl3
diff --git a/integration_test/cases/preload.exs b/integration_test/cases/preload.exs
index f5046799e0..11f2d06139 100644
--- a/integration_test/cases/preload.exs
+++ b/integration_test/cases/preload.exs
@@ -6,6 +6,7 @@ defmodule Ecto.Integration.PreloadTest do
 
   alias Ecto.Integration.Post
   alias Ecto.Integration.Comment
+  alias Ecto.Integration.CompositePk
   alias Ecto.Integration.Item
   alias Ecto.Integration.Permalink
   alias Ecto.Integration.User
@@ -347,6 +348,25 @@ defmodule Ecto.Integration.PreloadTest do
     assert [] = pe3.comments
   end
 
+  test "preload composite foreign key with function" do
+    c11 = TestRepo.insert!(%CompositePk{a: 1, b: 1, name: "11"})
+    c12 = TestRepo.insert!(%CompositePk{a: 1, b: 2, name: "12"})
+    c22 = TestRepo.insert!(%CompositePk{a: 2, b: 2, name: "22"})
+    c33 = TestRepo.insert!(%CompositePk{a: 3, b: 3, name: "33"})
+
+    TestRepo.insert!(%Post{title: "1", composite_a: 1, composite_b: 1})
+    TestRepo.insert!(%Post{title: "2", composite_a: 1, composite_b: 1})
+    TestRepo.insert!(%Post{title: "3", composite_a: 1, composite_b: 2})
+    TestRepo.insert!(%Post{title: "4", composite_a: 2, composite_b: 2})
+
+    assert [ce12, ce11, ce33, ce22] = TestRepo.preload([c12, c11, c33, c22],
+                                              posts: fn _ -> TestRepo.all(Post) end)
+    assert [%Post{title: "1"}, %Post{title: "2"}] = ce11.posts
+    assert [%Post{title: "3"}] = ce12.posts
+    assert [%Post{title: "4"}] = ce22.posts
+    assert [] = ce33.posts
+  end
+
   test "preload many_to_many with function" do
     p1 = TestRepo.insert!(%Post{title: "1"})
     p2 = TestRepo.insert!(%Post{title: "2"})
@@ -397,6 +417,52 @@ defmodule Ecto.Integration.PreloadTest do
     assert p3.users == [%{id: uid1}, %{id: uid4}]
   end
 
+  test "preload many_to_many on composite foreign keys with function" do
+    c11 = TestRepo.insert!(%CompositePk{a: 1, b: 1, name: "11"})
+    c12 = TestRepo.insert!(%CompositePk{a: 1, b: 2, name: "12"})
+    c22 = TestRepo.insert!(%CompositePk{a: 2, b: 2, name: "22"})
+
+    TestRepo.insert_all "composite_pk_composite_pk", [[a_1: 1, b_1: 1, a_2: 1, b_2: 2],
+                                                      [a_1: 1, b_1: 1, a_2: 2, b_2: 2],
+                                                      [a_1: 1, b_1: 2, a_2: 1, b_2: 1],
+                                                      [a_1: 2, b_1: 2, a_2: 2, b_2: 2]]
+
+    wrong_preloader = fn composite_ids ->
+      composite_ids_a = Enum.map(composite_ids, &Enum.at(&1, 0))
+      composite_ids_b = Enum.map(composite_ids, &Enum.at(&1, 1))
+      TestRepo.all(
+        from c in CompositePk,
+             join: cc in "composite_pk_composite_pk",
+             on: cc.a_2 == c.a and cc.b_2 == c.b,
+             where: cc.a_1 in ^composite_ids_a and cc.b_1 in ^composite_ids_b,
+             order_by: [c.a, c.b],
+             select: map(c, [:name])
+      )
+    end
+
+    assert_raise RuntimeError, ~r/invalid custom preload for `composites` on `Ecto.Integration.CompositePk`/, fn ->
+      TestRepo.preload([c11, c12, c22], composites: wrong_preloader)
+    end
+
+    right_preloader = fn composite_ids ->
+      composite_ids_a = Enum.map(composite_ids, &Enum.at(&1, 0))
+      composite_ids_b = Enum.map(composite_ids, &Enum.at(&1, 1))
+      TestRepo.all(
+        from c in CompositePk,
+             join: cc in "composite_pk_composite_pk",
+             on: cc.a_2 == c.a and cc.b_2 == c.b,
+             where: cc.a_1 in ^composite_ids_a and cc.b_1 in ^composite_ids_b,
+             order_by: [c.a, c.b],
+             select: {[cc.a_1, cc.b_1], map(c, [:name])}
+      )
+    end
+
+    [c11, c12, c22] = TestRepo.preload([c11, c12, c22], composites: right_preloader)
+    assert c11.composites == [%{name: "12"}, %{name: "22"}]
+    assert c12.composites == [%{name: "11"}]
+    assert c22.composites == [%{name: "22"}]
+  end
+
   test "preload with query" do
     p1 = TestRepo.insert!(%Post{title: "1"})
     p2 = TestRepo.insert!(%Post{title: "2"})
@@ -607,11 +673,23 @@ defmodule Ecto.Integration.PreloadTest do
 
     assert ExUnit.CaptureLog.capture_log(fn ->
       assert TestRepo.preload(updated, [:author]).author == u1
-    end) =~ ~r/its association key `author_id` is nil/
+    end) =~ ~r/its association keys `\(author_id\)` are nil/
 
     assert TestRepo.preload(updated, [:author], force: true).author == nil
   end
 
+  test "preload raises with association over composite foreign key is set but without id" do
+    p1 = TestRepo.insert!(%Post{title: "1"})
+    c11 = TestRepo.insert!(%CompositePk{a: 1, b: 1, name: "11"})
+    updated = %{p1 | composite: c11, composite_a: nil, composite_b: nil}
+
+    assert ExUnit.CaptureLog.capture_log(fn ->
+      assert TestRepo.preload(updated, [:composite]).composite == c11
+    end) =~ ~r/its association keys `\(composite_a, composite_b\)` are nil/
+
+    assert TestRepo.preload(updated, [:composite], force: true).composite == nil
+  end
+
   test "preload skips already loaded for cardinality one" do
     %Post{id: pid} = TestRepo.insert!(%Post{title: "1"})
 
diff --git a/integration_test/cases/repo.exs b/integration_test/cases/repo.exs
index 87c73063ae..91d8c8bd60 100644
--- a/integration_test/cases/repo.exs
+++ b/integration_test/cases/repo.exs
@@ -152,6 +152,23 @@ defmodule Ecto.Integration.RepoTest do
     assert TestRepo.all(PostUserCompositePk) == []
   end
 
+  @tag :composite_pk
+  test "insert, update and delete with assoc over composite foreign key" do
+    composite = TestRepo.insert!(%CompositePk{a: 1, b: 2, name: "name"})
+    post = TestRepo.insert!(%Post{title: "post title", composite: composite})
+
+    assert post.composite_a == 1
+    assert post.composite_b == 2
+    assert TestRepo.get_by!(CompositePk, [a: 1, b: 2]) == composite
+
+    post = post |> Ecto.Changeset.change(composite: nil) |> TestRepo.update!
+    assert is_nil(post.composite_a)
+    assert is_nil(post.composite_b)
+
+    TestRepo.delete!(post)
+    assert TestRepo.all(CompositePk) == [composite]
+  end
+
   @tag :invalid_prefix
   test "insert, update and delete with invalid prefix" do
     post = TestRepo.insert!(%Post{})
diff --git a/integration_test/support/schemas.exs b/integration_test/support/schemas.exs
index ea64651905..7e426484e0 100644
--- a/integration_test/support/schemas.exs
+++ b/integration_test/support/schemas.exs
@@ -54,6 +54,8 @@ defmodule Ecto.Integration.Post do
     has_one :update_permalink, Ecto.Integration.Permalink, foreign_key: :post_id, on_delete: :delete_all, on_replace: :update
     has_many :comments_authors, through: [:comments, :author]
     belongs_to :author, Ecto.Integration.User
+    belongs_to :composite, Ecto.Integration.CompositePk,
+      foreign_key: [:composite_a, :composite_b], references: [:a, :b], type: [:integer, :integer], on_replace: :nilify
     many_to_many :users, Ecto.Integration.User,
       join_through: "posts_users", on_delete: :delete_all, on_replace: :delete
     many_to_many :ordered_users, Ecto.Integration.User, join_through: "posts_users", preload_order: [desc: :name]
@@ -63,7 +65,7 @@ defmodule Ecto.Integration.Post do
       join_through: Ecto.Integration.PostUserCompositePk
     has_many :users_comments, through: [:users, :comments]
     has_many :comments_authors_permalinks, through: [:comments_authors, :permalink]
-    has_one :post_user_composite_pk, Ecto.Integration.PostUserCompositePk
+    has_many :post_user_composite_pk, Ecto.Integration.PostUserCompositePk
     timestamps()
   end
 
@@ -294,6 +296,12 @@ defmodule Ecto.Integration.CompositePk do
     field :a, :integer, primary_key: true
     field :b, :integer, primary_key: true
     field :name, :string
+    has_many :posts, Ecto.Integration.Post, foreign_key: [:composite_a, :composite_b], references: [:a, :b]
+    many_to_many :composites, Ecto.Integration.CompositePk,
+      join_through: "composite_pk_composite_pk", join_keys: [[a_1: :a, b_1: :b], [a_2: :a, b_2: :b]],
+      on_delete: :delete_all, on_replace: :delete
+    has_one :one_to_one_composite_pk, Ecto.Integration.OneToOneCompositePk,
+      foreign_key: [:composite_a, :composite_b], references: [:a, :b]
   end
   def changeset(schema, params) do
     cast(schema, params, ~w(a b name)a)
@@ -332,6 +340,23 @@ defmodule Ecto.Integration.PostUserCompositePk do
   end
 end
 
+defmodule Ecto.Integration.OneToOneCompositePk do
+  @moduledoc """
+  This module is used to test:
+
+    * Composite primary keys for 2 has_one fields
+
+  """
+  use Ecto.Integration.Schema
+
+  schema "one_to_one_composite_pk" do
+    belongs_to :composite, Ecto.Integration.CompositePk,
+      foreign_key: [:composite_a, :composite_b], references: [:a, :b], type: [:integer, :integer], on_replace: :nilify
+    timestamps()
+  end
+end
+
+
 defmodule Ecto.Integration.Usec do
   @moduledoc """
   This module is used to test:
diff --git a/lib/ecto.ex b/lib/ecto.ex
index 8ceed81ad4..ac0a09090f 100644
--- a/lib/ecto.ex
+++ b/lib/ecto.ex
@@ -523,10 +523,16 @@ defmodule Ecto do
     refl = %{owner_key: owner_key} = Ecto.Association.association_from_schema!(schema, assoc)
 
     values =
-      Enum.uniq for(struct <- structs,
-        assert_struct!(schema, struct),
-        key = Map.fetch!(struct, owner_key),
-        do: key)
+      structs
+      |> Enum.filter(&assert_struct!(schema, &1))
+      |> Enum.map(fn struct ->
+        case owner_key do
+          single_key when is_atom(single_key) -> Map.fetch!(struct, single_key)
+          [single_key] -> Map.fetch!(struct, single_key)
+          [_ | _] -> owner_key |> Enum.map(&Map.fetch!(struct, &1)) # |> List.to_tuple()
+        end
+      end)
+      |> Enum.uniq
 
     case assocs do
       [] ->
diff --git a/lib/ecto/association.ex b/lib/ecto/association.ex
index 1bf2e11d1f..3f645155d5 100644
--- a/lib/ecto/association.ex
+++ b/lib/ecto/association.ex
@@ -35,7 +35,7 @@ defmodule Ecto.Association do
                required(:cardinality) => :one | :many,
                required(:relationship) => :parent | :child,
                required(:owner) => atom,
-               required(:owner_key) => atom,
+               required(:owner_key) => list(atom),
                required(:field) => atom,
                required(:unique) => boolean,
                optional(atom) => any}
@@ -71,7 +71,8 @@ defmodule Ecto.Association do
 
     * `:owner` - the owner module of the association
 
-    * `:owner_key` - the key in the owner with the association value
+    * `:owner_key` - the key in the owner with the association value, or a
+      list of keys for composite keys
 
     * `:relationship` - if the relationship to the specified schema is
       of a `:child` or a `:parent`
@@ -235,8 +236,10 @@ defmodule Ecto.Association do
     # for the final WHERE clause with values.
     {_, query, _, dest_out_key} = Enum.reduce(joins, {source, query, counter, source.out_key}, fn curr_rel, {prev_rel, query, counter, _} ->
       related_queryable = curr_rel.schema
-
-      next = join(query, :inner, [{src, counter}], dest in ^related_queryable, on: field(src, ^prev_rel.out_key) == field(dest, ^curr_rel.in_key))
+      next = query
+        # join on the foreign key
+        |> join_on_keys(related_queryable, curr_rel.in_key, prev_rel.out_key, counter + 1, counter)
+        # consider where clauses on assocs
         |> combine_joins_query(curr_rel.where, counter + 1)
 
       {curr_rel, next, counter + 1, curr_rel.out_key}
@@ -245,22 +248,101 @@ defmodule Ecto.Association do
     final_bind = Ecto.Query.Builder.count_binds(query) - 1
 
     values = List.wrap(values)
-    query = case {join_to, values} do
-      {nil, [single_value]} ->
+
+    query = case {join_to, List.wrap(dest_out_key), values} do
+      {nil, [single_key], [single_value]} ->
         query
-        |> where([{dest, final_bind}], field(dest, ^dest_out_key) == ^single_value)
+        |> where([{dest, final_bind}], field(dest, ^single_key) == ^single_value)
 
-      {nil, values} ->
+      {nil, [single_key], values} ->
         query
-        |> where([{dest, final_bind}], field(dest, ^dest_out_key) in ^values)
+        |> where([{dest, final_bind}], field(dest, ^single_key) in ^values)
 
-      {_, _} ->
+      {nil, dest_out_keys, [single_value]} ->
+        dest_out_keys
+        |> Enum.zip(single_value)
+        |> Enum.reduce(query, fn {dest_out_key_field, value}, query ->
+          query
+          |> where([{dest, final_bind}], field(dest, ^dest_out_key_field) == ^value)
+        end)
+
+      {nil, dest_out_keys, values} ->
+        query
+        |> where_keys(final_bind, dest_out_keys, values)
+
+      {_, _, _} ->
         query
     end
 
     combine_assoc_query(query, source.where || [])
   end
 
+  @doc false
+  def join_on_keys(query, related_queryable, src_keys, dst_keys, src_binding, dst_binding) do
+    src_keys = List.wrap(src_keys)
+    dst_keys = List.wrap(dst_keys)
+    %{joins: joins} = query = join(query, :inner, [{src, src_binding}], dst in ^related_queryable, on: true)
+    {%{on: %{expr: _expr} = on} = last_join, joins} = joins |> List.pop_at(-1)
+
+    expr = Enum.reduce strict_zip(src_keys, dst_keys), true, fn {src_key, dst_key}, expr ->
+      conjoin_exprs(expr, {:==, [], [to_field(dst_binding, dst_key), to_field(src_binding, src_key)]})
+    end
+
+    %{query | joins: joins ++ [%{last_join | on: %{on | expr: expr}}]}
+  end
+
+  @doc false
+  def strict_zip([l | ls], [r | rs]), do: [{l, r} | strict_zip(ls, rs)]
+  def strict_zip([], []), do: []
+  def strict_zip(_, _), do: raise ArgumentError, "lists should be of equal length"
+
+  @doc false
+  def conjoin_exprs(true, r), do: r
+  def conjoin_exprs(l, r), do: {:and, [], [l, r]}
+
+  def where_keys(%{wheres: wheres} = query, binding, keys, values) do
+    keys = List.wrap(keys)
+    values = List.wrap(values)
+    {expr, params, op} = where_expr(keys, values, binding)
+    %{query | wheres: wheres ++ [%Ecto.Query.BooleanExpr{op: op, expr: expr, params: params, line: __ENV__.line, file: __ENV__.file}]}
+  end
+
+  defp where_expr([key], [nil], binding) do
+    expr = {:is_nil, [], [to_field(binding, key)]}
+    {expr, [], :and}
+  end
+
+  defp where_expr([key], [value], binding) do
+    expr = {:==, [], [to_field(binding, key), {:^, [], [0]}]}
+    params = [{value, {binding, key}}]
+    {expr, params, :and}
+  end
+
+  defp where_expr([key], values, binding) do
+    expr = {:in, [], [to_field(binding, key), {:^, [], [0]}]}
+    params = [{values, {:in, {binding, key}}}]
+    {expr, params, :and}
+  end
+
+  defp where_expr(keys, values, binding) do
+    or_exprs = fn
+      false, r -> r
+      l, r -> {:or, [], [l, r]}
+    end
+
+    {expr, params, _counter} =
+      Enum.reduce(values, {false, [], 0}, fn values, {expr, params, counter} ->
+        {new_expr, params, counter} = strict_zip(keys, values)
+          |> Enum.reduce({true, params, counter}, fn {key, value}, {expr, params, counter} ->
+            expr = conjoin_exprs(expr, {:==, [], [to_field(binding, key), {:^, [], [counter]}]})
+            {expr, [{value, {binding, key}} | params], counter + 1}
+          end)
+        {or_exprs.(expr, new_expr), params, counter}
+      end)
+
+    {expr, Enum.reverse(params), :or}
+  end
+
   defp flatten_through_chain(owner, [], acc), do: {owner, acc}
   defp flatten_through_chain(owner, [assoc | tl], acc) do
     refl = association_from_schema!(owner, assoc)
@@ -290,11 +372,16 @@ defmodule Ecto.Association do
 
       table_list = case refl do
         %{join_through: join_through, join_keys: join_keys, join_where: join_where, where: where} ->
-          [{owner_join_key, owner_key}, {related_join_key, related_key}] = join_keys
+          %{
+            owner_keys: owner_keys,
+            owner_join_keys: owner_join_keys,
+            related_keys: related_keys,
+            related_join_keys: related_join_keys
+          } = resolve_join_keys(join_keys)
 
-          owner_map = %{owner_map | in_key: owner_key}
-          join_map = %{schema: join_through, out_key: owner_join_key, in_key: related_join_key, where: join_where}
-          related_map = %{schema: refl.related, out_key: related_key, in_key: nil, where: where}
+          owner_map = %{owner_map | in_key: owner_keys}
+          join_map = %{schema: join_through, out_key: owner_join_keys, in_key: related_join_keys, where: join_where}
+          related_map = %{schema: refl.related, out_key: related_keys, in_key: nil, where: where}
 
           [related_map, join_map, owner_map | table_list]
 
@@ -320,6 +407,15 @@ defmodule Ecto.Association do
     end)
   end
 
+  defp resolve_join_keys([owner_join_keys, related_join_keys]) do
+    owner_keys = Keyword.values(owner_join_keys)
+    owner_join_keys = Keyword.keys(owner_join_keys)
+
+    related_keys = Keyword.values(related_join_keys)
+    related_join_keys = Keyword.keys(related_join_keys)
+    %{owner_keys: owner_keys, owner_join_keys: owner_join_keys, related_keys: related_keys, related_join_keys: related_join_keys}
+  end
+
   @doc """
   Add the default assoc query where clauses to a join.
 
@@ -351,39 +447,35 @@ defmodule Ecto.Association do
   end
 
   defp expand_where(conditions, expr, params, counter, binding) do
-    conjoin_exprs = fn
-      true, r -> r
-      l, r-> {:and, [], [l, r]}
-    end
-
     {expr, params, _counter} =
       Enum.reduce(conditions, {expr, params, counter}, fn
         {key, nil}, {expr, params, counter} ->
-          expr = conjoin_exprs.(expr, {:is_nil, [], [to_field(binding, key)]})
+          expr = conjoin_exprs(expr, {:is_nil, [], [to_field(binding, key)]})
           {expr, params, counter}
 
         {key, {:not, nil}}, {expr, params, counter} ->
-          expr = conjoin_exprs.(expr, {:not, [], [{:is_nil, [], [to_field(binding, key)]}]})
+          expr = conjoin_exprs(expr, {:not, [], [{:is_nil, [], [to_field(binding, key)]}]})
           {expr, params, counter}
 
         {key, {:fragment, frag}}, {expr, params, counter} when is_binary(frag) ->
           pieces = Ecto.Query.Builder.fragment_pieces(frag, [to_field(binding, key)])
-          expr = conjoin_exprs.(expr, {:fragment, [], pieces})
+          expr = conjoin_exprs(expr, {:fragment, [], pieces})
           {expr, params, counter}
 
         {key, {:in, value}}, {expr, params, counter} when is_list(value) ->
-          expr = conjoin_exprs.(expr, {:in, [], [to_field(binding, key), {:^, [], [counter]}]})
+          expr = conjoin_exprs(expr, {:in, [], [to_field(binding, key), {:^, [], [counter]}]})
           {expr, [{value, {:in, {binding, key}}} | params], counter + 1}
 
         {key, value}, {expr, params, counter} ->
-          expr = conjoin_exprs.(expr, {:==, [], [to_field(binding, key), {:^, [], [counter]}]})
+          expr = conjoin_exprs(expr, {:==, [], [to_field(binding, key), {:^, [], [counter]}]})
           {expr, [{value, {binding, key}} | params], counter + 1}
       end)
 
     {expr, Enum.reverse(params)}
   end
 
-  defp to_field(binding, field),
+  @doc false
+  def to_field(binding, field),
     do: {{:., [], [{:&, [], [binding]}, field]}, [], []}
 
   @doc """
@@ -633,6 +725,22 @@ defmodule Ecto.Association do
 
   defp primary_key!(nil), do: []
   defp primary_key!(struct), do: Ecto.primary_key!(struct)
+
+  def missing_fields(queryable, related_key) do
+    related_key
+    |> List.wrap()
+    |> Enum.filter(&is_nil(queryable.__schema__(:type, &1)))
+  end
+
+  def missing_primary_keys(queryable, related_key) do
+    related_key = List.wrap related_key
+    Enum.reject queryable.__schema__(:primary_key), &(&1 in related_key)
+  end
+
+  def label(env) do
+    {fun, arity} = env.function
+    "#{env.module}.#{fun}/#{arity} #{env.line}"
+  end
 end
 
 defmodule Ecto.Association.Has do
@@ -645,8 +753,8 @@ defmodule Ecto.Association.Has do
     * `field` - The name of the association field on the schema
     * `owner` - The schema where the association was defined
     * `related` - The schema that is associated
-    * `owner_key` - The key on the `owner` schema used for the association
-    * `related_key` - The key on the `related` schema used for the association
+    * `owner_key` - The key or list of columns that form the key on the `owner` schema used for the association
+    * `related_key` - The key or list of columns that form the key on the `related` schema used for the association
     * `queryable` - The real query to use for querying association
     * `on_delete` - The action taken on associations when schema is deleted
     * `on_replace` - The action taken on associations when schema is replaced
@@ -674,27 +782,29 @@ defmodule Ecto.Association.Has do
         {:error, "associated schema #{inspect queryable} does not exist"}
       not function_exported?(queryable, :__schema__, 2) ->
         {:error, "associated module #{inspect queryable} is not an Ecto schema"}
-      is_nil queryable.__schema__(:type, related_key) ->
-        {:error, "associated schema #{inspect queryable} does not have field `#{related_key}`"}
+      [] != (missing_fields = Ecto.Association.missing_fields(queryable, related_key)) ->
+        {:error, "associated schema #{inspect queryable} does not have field(s) `#{inspect missing_fields}`"}
       true ->
         :ok
     end
   end
 
   @impl true
-  def struct(module, name, opts) do
+  def struct(module, name, opts) do 
     queryable = Keyword.fetch!(opts, :queryable)
     cardinality = Keyword.fetch!(opts, :cardinality)
     related = Ecto.Association.related_from_query(queryable, name)
 
     ref =
       module
-      |> Module.get_attribute(:primary_key)
+      |> Module.get_attribute(:ecto_primary_keys)
       |> get_ref(opts[:references], name)
 
-    unless Module.get_attribute(module, :ecto_fields)[ref] do
-      raise ArgumentError, "schema does not have the field #{inspect ref} used by " <>
-        "association #{inspect name}, please set the :references option accordingly"
+    for ref <- List.wrap(ref) do
+      unless Module.get_attribute(module, :ecto_fields)[ref] do
+        raise ArgumentError, "schema does not have the field #{inspect ref} used by " <>
+          "association #{inspect name}, please set the :references option accordingly"
+      end
     end
 
     if opts[:through] do
@@ -726,13 +836,19 @@ defmodule Ecto.Association.Has do
       raise ArgumentError, "expected `:where` for #{inspect name} to be a keyword list, got: `#{inspect where}`"
     end
 
+    foreign_key = case opts[:foreign_key] do
+      nil when is_atom(ref) -> Ecto.Association.association_key(module, ref)
+      nil when is_list(ref) -> Enum.map(ref, &Ecto.Association.association_key(module, &1))
+      key_or_keys  -> key_or_keys
+    end
+
     %__MODULE__{
       field: name,
       cardinality: cardinality,
       owner: module,
       related: related,
       owner_key: ref,
-      related_key: opts[:foreign_key] || Ecto.Association.association_key(module, ref),
+      related_key: foreign_key,
       queryable: queryable,
       on_delete: on_delete,
       on_replace: on_replace,
@@ -742,12 +858,14 @@ defmodule Ecto.Association.Has do
     }
   end
 
-  defp get_ref(primary_key, nil, name) when primary_key in [nil, false] do
+  defp get_ref([] = _primary_keys, nil, name) do
     raise ArgumentError, "need to set :references option for " <>
       "association #{inspect name} when schema has no primary key"
   end
-  defp get_ref(primary_key, nil, _name), do: elem(primary_key, 0)
-  defp get_ref(_primary_key, references, _name), do: references
+  defp get_ref([primary_key], nil, _name), do: primary_key
+  defp get_ref(primary_keys, nil, _name), do: Enum.reverse(primary_keys)
+  defp get_ref(_primary_keys, [reference], _name), do: reference
+  defp get_ref(_primary_keys, references, _name), do: references
 
   @impl true
   def build(%{owner_key: owner_key, related_key: related_key} = refl, owner, attributes) do
@@ -757,19 +875,15 @@ defmodule Ecto.Association.Has do
 
   @impl true
   def joins_query(%{related_key: related_key, owner: owner, owner_key: owner_key, queryable: queryable} = assoc) do
-    from(o in owner, join: q in ^queryable, on: field(q, ^related_key) == field(o, ^owner_key))
+    from(o in owner)
+    |> Ecto.Association.join_on_keys(queryable, owner_key, related_key, 0, 1)
     |> Ecto.Association.combine_joins_query(assoc.where, 1)
   end
 
-  @impl true
-  def assoc_query(%{related_key: related_key, queryable: queryable} = assoc, query, [value]) do
-    from(x in (query || queryable), where: field(x, ^related_key) == ^value)
-    |> Ecto.Association.combine_assoc_query(assoc.where)
-  end
-
   @impl true
   def assoc_query(%{related_key: related_key, queryable: queryable} = assoc, query, values) do
-    from(x in (query || queryable), where: field(x, ^related_key) in ^values)
+    from(x in (query || queryable))
+    |> Ecto.Association.where_keys(0, related_key, values)
     |> Ecto.Association.combine_assoc_query(assoc.where)
   end
 
@@ -808,16 +922,21 @@ defmodule Ecto.Association.Has do
     %{data: parent, repo: repo} = parent_changeset
     %{action: action, changes: changes} = changeset
 
-    {key, value} = parent_key(assoc, parent)
-    changeset = update_parent_key(changeset, action, key, value)
-    changeset = Ecto.Association.update_parent_prefix(changeset, parent)
+    parent_keys = parent_keys(assoc, parent)
+    changeset = Enum.reduce parent_keys, changeset, fn {key, value}, changeset ->
+      changeset = update_parent_key(changeset, action, key, value)
+      Ecto.Association.update_parent_prefix(changeset, parent)
+    end
 
     case apply(repo, action, [changeset, opts]) do
       {:ok, _} = ok ->
         if action == :delete, do: {:ok, nil}, else: ok
       {:error, changeset} ->
-        original = Map.get(changes, key)
-        {:error, put_in(changeset.changes[key], original)}
+        changeset = Enum.reduce parent_keys, changeset, fn {key, _}, changeset ->
+          original = Map.get(changes, key)
+          put_in(changeset.changes[key], original)
+        end
+        {:error, changeset}
     end
   end
 
@@ -826,11 +945,21 @@ defmodule Ecto.Association.Has do
   defp update_parent_key(changeset, _action, key, value),
     do: Ecto.Changeset.put_change(changeset, key, value)
 
-  defp parent_key(%{related_key: related_key}, nil) do
-    {related_key, nil}
+  defp parent_keys(%{related_key: related_keys}, nil) when is_list(related_keys) do
+    Enum.map related_keys, fn related_key -> {related_key, nil} end
+  end
+  defp parent_keys(%{related_key: related_key}, nil) do
+    [{related_key, nil}]
+  end
+  defp parent_keys(%{owner_key: owner_keys, related_key: related_keys}, owner) when is_list(owner_keys) and is_list(related_keys) do
+    owner_keys
+    |> Enum.zip(related_keys)
+    |> Enum.map(fn {owner_key, related_key} ->
+      {related_key, Map.get(owner, owner_key)}
+    end)
   end
-  defp parent_key(%{owner_key: owner_key, related_key: related_key}, owner) do
-    {related_key, Map.get(owner, owner_key)}
+  defp parent_keys(%{owner_key: owner_key, related_key: related_key}, owner) do
+    [{related_key, Map.get(owner, owner_key)}]
   end
 
   ## Relation callbacks
@@ -853,18 +982,35 @@ defmodule Ecto.Association.Has do
   end
 
   @doc false
-  def nilify_all(%{related_key: related_key} = refl, parent, repo_name, opts) do
+  def nilify_all(%{related_key: related_key} = refl, parent, repo_name, opts) when is_atom(related_key) do
     if query = on_delete_query(refl, parent) do
       Ecto.Repo.Queryable.update_all repo_name, query, [set: [{related_key, nil}]], opts
     end
   end
+  def nilify_all(%{related_key: related_keys} = refl, parent, repo_name, opts) do
+    if query = on_delete_query(refl, parent) do
+      set = Enum.map(related_keys, fn related_key -> {related_key, nil} end)
+      Ecto.Repo.Queryable.update_all repo_name, query, [set: set], opts
+    end
+  end
 
-  defp on_delete_query(%{owner_key: owner_key, related_key: related_key,
-                         queryable: queryable}, parent) do
+  defp on_delete_query(%{owner_key: owner_key, related_key: related_key, queryable: queryable}, parent)
+                       when is_atom(owner_key) and is_atom(related_key) do
     if value = Map.get(parent, owner_key) do
       from x in queryable, where: field(x, ^related_key) == ^value
     end
   end
+
+  defp on_delete_query(%{owner_key: owner_keys, related_key: related_keys, queryable: queryable}, parent) do
+    values = Enum.map(owner_keys, &Map.get(parent, &1))
+    if values != [] && Enum.all?(values) do
+      related_keys
+      |> Enum.zip(values)
+      |> Enum.reduce(from(x in queryable), fn {related_key_field, value}, query ->
+        query |> where([x], field(x, ^related_key_field) == ^ value)
+      end)
+    end
+  end
 end
 
 defmodule Ecto.Association.HasThrough do
@@ -876,7 +1022,7 @@ defmodule Ecto.Association.HasThrough do
     * `cardinality` - The association cardinality
     * `field` - The name of the association field on the schema
     * `owner` - The schema where the association was defined
-    * `owner_key` - The key on the `owner` schema used for the association
+    * `owner_key` - The column or list of columns that form the key on the `owner` schema used for the association
     * `through` - The through associations
     * `relationship` - The relationship to the specified schema, default `:child`
   """
@@ -943,7 +1089,7 @@ defmodule Ecto.Association.HasThrough do
   end
 
   @impl true
-  def assoc_query(%{owner: owner, through: through}, _, values) do
+  def assoc_query(%{owner: owner, through: through}, _query, values) do
     Ecto.Association.filter_through_chain(owner, through, values)
   end
 end
@@ -957,9 +1103,9 @@ defmodule Ecto.Association.BelongsTo do
     * `cardinality` - The association cardinality
     * `field` - The name of the association field on the schema
     * `owner` - The schema where the association was defined
-    * `owner_key` - The key on the `owner` schema used for the association
+    * `owner_key` - The key or list of keys on the `owner` schema used for the association
     * `related` - The schema that is associated
-    * `related_key` - The key on the `related` schema used for the association
+    * `related_key` - The key or list of keys on the `related` schema used for the association
     * `queryable` - The real query to use for querying association
     * `defaults` - Default fields used when building the association
     * `relationship` - The relationship to the specified schema, default `:parent`
@@ -983,8 +1129,10 @@ defmodule Ecto.Association.BelongsTo do
         {:error, "associated schema #{inspect queryable} does not exist"}
       not function_exported?(queryable, :__schema__, 2) ->
         {:error, "associated module #{inspect queryable} is not an Ecto schema"}
-      is_nil queryable.__schema__(:type, related_key) ->
-        {:error, "associated schema #{inspect queryable} does not have field `#{related_key}`"}
+      [] != (missing_fields = Ecto.Association.missing_fields(queryable, related_key)) ->
+        {:error, "associated schema #{inspect queryable} does not have field(s) `#{inspect missing_fields}`"}
+      [] != (missing_pks = Ecto.Association.missing_primary_keys(queryable, related_key)) ->
+        {:error, "associated schema #{inspect queryable} has primary keys #{inspect missing_pks} not included in association"}
       true ->
         :ok
     end
@@ -1032,19 +1180,15 @@ defmodule Ecto.Association.BelongsTo do
 
   @impl true
   def joins_query(%{related_key: related_key, owner: owner, owner_key: owner_key, queryable: queryable} = assoc) do
-    from(o in owner, join: q in ^queryable, on: field(q, ^related_key) == field(o, ^owner_key))
+    from(o in owner)
+    |> Ecto.Association.join_on_keys(queryable, owner_key, related_key, 0, 1)
     |> Ecto.Association.combine_joins_query(assoc.where, 1)
   end
 
-  @impl true
-  def assoc_query(%{related_key: related_key, queryable: queryable} = assoc, query, [value]) do
-    from(x in (query || queryable), where: field(x, ^related_key) == ^value)
-    |> Ecto.Association.combine_assoc_query(assoc.where)
-  end
-
   @impl true
   def assoc_query(%{related_key: related_key, queryable: queryable} = assoc, query, values) do
-    from(x in (query || queryable), where: field(x, ^related_key) in ^values)
+    from(x in (query || queryable))
+    |> Ecto.Association.where_keys(0, related_key, values)
     |> Ecto.Association.combine_assoc_query(assoc.where)
   end
 
@@ -1110,7 +1254,7 @@ defmodule Ecto.Association.ManyToMany do
     * `field` - The name of the association field on the schema
     * `owner` - The schema where the association was defined
     * `related` - The schema that is associated
-    * `owner_key` - The key on the `owner` schema used for the association
+    * `owner_key` - The list of fields that form the key on the `owner` schema used for the association
     * `queryable` - The real query to use for querying association
     * `on_delete` - The action taken on associations when schema is deleted
     * `on_replace` - The action taken on associations when schema is replaced
@@ -1161,16 +1305,20 @@ defmodule Ecto.Association.ManyToMany do
 
     join_keys = opts[:join_keys]
     join_through = opts[:join_through]
-    validate_join_through(name, join_through)
+    validate_join_through!(name, join_through)
 
     {owner_key, join_keys} =
       case join_keys do
         [{join_owner_key, owner_key}, {join_related_key, related_key}]
             when is_atom(join_owner_key) and is_atom(owner_key) and
                  is_atom(join_related_key) and is_atom(related_key) ->
-          {owner_key, join_keys}
+          join_keys = [[{join_owner_key, owner_key}], [{join_related_key, related_key}]]
+          {[owner_key], join_keys}
+        [join_keys_to_through, join_keys_to_related]
+            when is_list(join_keys_to_through) and is_list(join_keys_to_related) ->
+          {Keyword.values(join_keys_to_through), join_keys}
         nil ->
-          {:id, default_join_keys(module, related)}
+          {[:id], default_join_keys(module, related)}
         _ ->
           raise ArgumentError,
             "many_to_many #{inspect name} expect :join_keys to be a keyword list " <>
@@ -1179,10 +1327,12 @@ defmodule Ecto.Association.ManyToMany do
             "the associated schema. For example: #{inspect default_join_keys(module, related)}"
       end
 
-    unless Module.get_attribute(module, :ecto_fields)[owner_key] do
-      raise ArgumentError, "schema does not have the field #{inspect owner_key} used by " <>
-        "association #{inspect name}, please set the :join_keys option accordingly"
-    end
+    Enum.each(owner_key, fn owner_key_field ->
+      unless Module.get_attribute(module, :ecto_fields)[owner_key_field] do
+        raise ArgumentError, "schema does not have the field #{inspect owner_key_field} used by " <>
+          "association #{inspect name}, please set the :join_keys option accordingly"
+      end
+    end)
 
     on_delete  = Keyword.get(opts, :on_delete, :nothing)
     on_replace = Keyword.get(opts, :on_replace, :raise)
@@ -1238,18 +1388,20 @@ defmodule Ecto.Association.ManyToMany do
   end
 
   defp default_join_keys(module, related) do
-    [{Ecto.Association.association_key(module, :id), :id},
-     {Ecto.Association.association_key(related, :id), :id}]
+    [
+      [{Ecto.Association.association_key(module, :id), :id}],
+      [{Ecto.Association.association_key(related, :id), :id}]
+    ]
   end
 
   @impl true
   def joins_query(%{owner: owner, queryable: queryable,
                     join_through: join_through, join_keys: join_keys} = assoc) do
-    [{join_owner_key, owner_key}, {join_related_key, related_key}] = join_keys
+    [join_through_keys, join_related_keys] = join_keys
 
-    from(o in owner,
-      join: j in ^join_through, on: field(j, ^join_owner_key) == field(o, ^owner_key),
-      join: q in ^queryable, on: field(j, ^join_related_key) == field(q, ^related_key))
+    from(o in owner)
+    |> Ecto.Association.join_on_keys(join_through, Keyword.keys(join_through_keys), Keyword.values(join_through_keys), 1, 0)
+    |> Ecto.Association.join_on_keys(queryable, Keyword.values(join_related_keys), Keyword.keys(join_related_keys), 2, 1)
     |> Ecto.Association.combine_joins_query(assoc.where, 2)
     |> Ecto.Association.combine_joins_query(assoc.join_where, 1)
   end
@@ -1261,17 +1413,21 @@ defmodule Ecto.Association.ManyToMany do
   @impl true
   def assoc_query(assoc, query, values) do
     %{queryable: queryable, join_through: join_through, join_keys: join_keys, owner: owner} = assoc
-    [{join_owner_key, owner_key}, {join_related_key, related_key}] = join_keys
-
-    owner_key_type = owner.__schema__(:type, owner_key)
+    [join_through_keys, join_related_keys] = join_keys
 
     # We only need to join in the "join table". Preload and Ecto.assoc expressions can then filter
     # by &1.join_owner_key in ^... to filter down to the associated entries in the related table.
-    query =
-      from(q in (query || queryable),
-        join: j in ^join_through, on: field(q, ^related_key) == field(j, ^join_related_key),
-        where: field(j, ^join_owner_key) in type(^values, {:in, ^owner_key_type})
-      )
+
+    dst_binding = if is_nil(query) do 
+      1
+    else
+      Ecto.Query.Builder.count_binds(query)
+    end
+
+    query = 
+      from(q in (query || queryable))
+      |> Ecto.Association.join_on_keys(join_through, Keyword.keys(join_related_keys), Keyword.values(join_related_keys), dst_binding, 0)
+      |> where_keys(owner, 1, join_through_keys, values)
       |> Ecto.Association.combine_assoc_query(assoc.where)
 
     Ecto.Association.combine_joins_query(query, assoc.join_where, length(query.joins))
@@ -1285,12 +1441,13 @@ defmodule Ecto.Association.ManyToMany do
   end
 
   @impl true
-  def preload_info(%{join_keys: [{join_owner_key, owner_key}, {_, _}], owner: owner} = refl) do
-    owner_key_type = owner.__schema__(:type, owner_key)
+  def preload_info(%{join_keys: [join_through_keys, _], owner: owner} = refl) do
+    join_owner_keys = Keyword.keys(join_through_keys)
+    owner_key_types = join_through_keys |> Keyword.values |> Enum.map(& owner.__schema__(:type, &1))
 
     # When preloading use the last bound table (which is the join table) and the join_owner_key
     # to filter out related entities to the owner structs we're preloading with.
-    {:assoc, refl, {-1, join_owner_key, owner_key_type}}
+    {:assoc, refl, {-1, join_owner_keys, owner_key_types}}
   end
 
   @impl true
@@ -1301,15 +1458,20 @@ defmodule Ecto.Association.ManyToMany do
 
   def on_repo_change(%{join_keys: join_keys, join_through: join_through, join_where: join_where},
                      %{repo: repo, data: owner}, %{action: :delete, data: related}, adapter, opts) do
-    [{join_owner_key, owner_key}, {join_related_key, related_key}] = join_keys
-    owner_value = dump! :delete, join_through, owner, owner_key, adapter
-    related_value = dump! :delete, join_through, related, related_key, adapter
+    [join_through_keys, join_related_keys] = join_keys
+    owner_values = dump! :delete, join_through, owner, Keyword.values(join_through_keys), adapter
+    related_values = dump! :delete, join_through, related, Keyword.values(join_related_keys), adapter
 
-    query =
-      join_through
-      |> where([j], field(j, ^join_owner_key) == ^owner_value)
-      |> where([j], field(j, ^join_related_key) == ^related_value)
-      |> Ecto.Association.combine_assoc_query(join_where)
+
+    join_fields = Keyword.keys(join_through_keys) ++ Keyword.keys(join_related_keys)
+    join_values = owner_values ++ related_values
+
+    query = join_fields
+            |> Enum.zip(join_values)
+            |> Enum.reduce(from(j in join_through), fn {field, value}, query ->
+              where(query, [j], field(j, ^field) == ^value)
+            end)
+            |> Ecto.Association.combine_assoc_query(join_where)
 
     query = %{query | prefix: owner.__meta__.prefix}
     repo.delete_all(query, opts)
@@ -1323,12 +1485,17 @@ defmodule Ecto.Association.ManyToMany do
 
     case apply(repo, action, [changeset, opts]) do
       {:ok, related} ->
-        [{join_owner_key, owner_key}, {join_related_key, related_key}] = join_keys
-
-        if insert_join?(parent_changeset, changeset, field, related_key) do
-          owner_value = dump! :insert, join_through, owner, owner_key, adapter
-          related_value = dump! :insert, join_through, related, related_key, adapter
-          data = %{join_owner_key => owner_value, join_related_key => related_value}
+        [join_through_keys, join_related_keys] = join_keys
+        owner_keys = Keyword.values(join_through_keys)
+        related_keys = Keyword.values(join_related_keys)
+
+        if insert_join?(parent_changeset, changeset, field, related_keys) do
+          owner_values = dump! :insert, join_through, owner, owner_keys, adapter
+          related_values = dump! :insert, join_through, related, related_keys, adapter
+          data =
+            (Keyword.keys(join_through_keys) ++ Keyword.keys(join_related_keys))
+            |> Enum.zip(owner_values ++ related_values)
+            |> Map.new
 
           case insert_join(join_through, refl, parent_changeset, data, opts) do
             {:error, join_changeset} ->
@@ -1346,24 +1513,26 @@ defmodule Ecto.Association.ManyToMany do
     end
   end
 
-  defp validate_join_through(name, nil) do
+  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(_, join_through) when is_atom(join_through) or is_binary(join_through) do
+  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
+  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) or a string (representing a table)"
   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
-    current_key = Map.fetch!(related, related_key)
-    not Enum.any? Map.fetch!(owner, field), fn child ->
-      Map.get(child, related_key) == current_key
+  defp insert_join?(%{action: :insert}, _, _field, _related_keys), do: true
+  defp insert_join?(_, %{action: :insert}, _field, _related_keys), do: true
+  defp insert_join?(%{data: owner}, %{data: related}, field, related_keys) do
+    Enum.all? related_keys, fn related_key ->
+      current_key = Map.fetch!(related, related_key)
+      not Enum.any? Map.fetch!(owner, field), fn child ->
+        Map.get(child, related_key) == current_key
+      end
     end
   end
 
@@ -1399,6 +1568,10 @@ defmodule Ecto.Association.ManyToMany do
     Map.get(struct, field) || raise "could not #{op} join entry because `#{field}` is nil in #{inspect struct}"
   end
 
+  defp dump!(action, join_through, struct, fields, adapter) when is_list(fields) do
+    Enum.map(fields, &dump!(action, join_through, struct, &1, adapter))
+  end
+
   defp dump!(action, join_through, struct, field, adapter) when is_binary(join_through) do
     value = field!(action, struct, field)
     type  = struct.__struct__.__schema__(:type, field)
@@ -1432,12 +1605,55 @@ defmodule Ecto.Association.ManyToMany do
   @doc false
   def delete_all(refl, parent, repo_name, opts) do
     %{join_through: join_through, join_keys: join_keys, owner: owner} = refl
-    [{join_owner_key, owner_key}, {_, _}] = join_keys
+    [join_through_keys, _join_related_keys] = join_keys
+    values = join_through_keys |> Keyword.values() |> Enum.map(&Map.get(parent, &1))
 
-    if value = Map.get(parent, owner_key) do
-      owner_type = owner.__schema__(:type, owner_key)
-      query = from j in join_through, where: field(j, ^join_owner_key) == type(^value, ^owner_type)
+
+    unless Enum.all?(values, &is_nil/1) do
+      query = from(j in join_through) |> where_keys(owner, 0, join_through_keys, [values])
       Ecto.Repo.Queryable.delete_all repo_name, query, opts
     end
   end
+
+  defp where_keys(%{wheres: wheres} = query, owner, binding, keys, values) do
+    keys = List.wrap(keys)
+    values = List.wrap(values)
+    {expr, params, op} = where_expr(owner, keys, values, binding)
+    %{query | wheres: wheres ++ [%Ecto.Query.BooleanExpr{op: op, expr: expr, params: params, line: __ENV__.line, file: __ENV__.file}]}
+  end
+
+  defp where_expr(owner, [{join_owner_key, owner_key}], [[value]], binding) do
+    owner_type = owner.__schema__(:type, owner_key)
+    expr = {:==, [], [Ecto.Association.to_field(binding, join_owner_key), {:type, [], [{:^, [], [0]}, owner_type]}]}
+    params = [{value, owner_type}]
+    {expr, params, :and}
+  end
+
+  defp where_expr(owner, [{join_owner_key, owner_key}], values, binding) do
+    owner_type = owner.__schema__(:type, owner_key)
+    # expr = {:in, [], [Ecto.Association.to_field(binding, join_owner_key), {:type, [], [{:^, [], [0]}, owner_type]}]}
+    expr = {:in, [], [Ecto.Association.to_field(binding, join_owner_key), {:^, [], [0]}]}
+    params = [{List.flatten(values), {:in, owner_type}}]
+    {expr, params, :and}
+  end
+
+  defp where_expr(owner, keys, values, binding) do
+    or_exprs = fn
+      false, r -> r
+      l, r -> {:or, [], [l, r]}
+    end
+
+    {expr, params, _counter} =
+      Enum.reduce(values, {false, [], 0}, fn values, {expr, params, counter} ->
+        {new_expr, params, counter} = Ecto.Association.strict_zip(keys, values)
+          |> Enum.reduce({true, params, counter}, fn {{join_owner_key, owner_key}, value}, {expr, params, counter} ->
+            owner_type = owner.__schema__(:type, owner_key)
+            expr = Ecto.Association.conjoin_exprs(expr, {:==, [], [Ecto.Association.to_field(binding, join_owner_key), {:type, [], [{:^, [], [counter]}, owner_type]}]})
+            {expr, [{value, owner_type} | params], counter + 1}
+          end)
+        {or_exprs.(expr, new_expr), params, counter}
+      end)
+
+    {expr, Enum.reverse(params), :and}
+  end
 end
diff --git a/lib/ecto/changeset.ex b/lib/ecto/changeset.ex
index e6b0f207e4..50d5103a1f 100644
--- a/lib/ecto/changeset.ex
+++ b/lib/ecto/changeset.ex
@@ -3506,7 +3506,7 @@ defmodule Ecto.Changeset do
     name = opts[:name] ||
       case get_assoc_type(changeset, assoc) do
         %Ecto.Association.BelongsTo{owner_key: owner_key} ->
-          "#{get_source(changeset)}_#{owner_key}_fkey"
+          "#{get_source(changeset)}_#{atom_concat owner_key}_fkey"
         other ->
           raise ArgumentError,
             "assoc_constraint can only be added to belongs to associations, got: #{inspect other}"
@@ -3564,7 +3564,7 @@ defmodule Ecto.Changeset do
       case get_assoc_type(changeset, assoc) do
         %Ecto.Association.Has{cardinality: cardinality,
                               related_key: related_key, related: related} ->
-          {opts[:name] || "#{related.__schema__(:source)}_#{related_key}_fkey",
+          {opts[:name] || "#{related.__schema__(:source)}_#{atom_concat related_key}_fkey",
            message(opts, no_assoc_message(cardinality))}
         other ->
           raise ArgumentError,
@@ -3799,6 +3799,13 @@ defmodule Ecto.Changeset do
     |> merge_keyword_keys(msg_func, changeset)
     |> merge_related_keys(changes, types, msg_func, &traverse_validations/2)
   end
+
+  defp atom_concat(atom) when is_atom(atom), do: Atom.to_string(atom)
+  defp atom_concat(atoms) when is_list(atoms) do
+    atoms
+    |> Enum.map(&Atom.to_string/1)
+    |> Enum.join("_")
+  end
 end
 
 defimpl Inspect, for: Ecto.Changeset do
diff --git a/lib/ecto/query/planner.ex b/lib/ecto/query/planner.ex
index b7f5284e8c..2cac6f0e0f 100644
--- a/lib/ecto/query/planner.ex
+++ b/lib/ecto/query/planner.ex
@@ -1859,7 +1859,7 @@ defmodule Ecto.Query.Planner do
         case schema.__schema__(:association, field) do
           %Ecto.Association.BelongsTo{owner_key: owner_key} ->
             error! query, expr, "field `#{field}` in `#{kind}` is an association in schema #{inspect schema}. " <>
-                                "Did you mean to use `#{owner_key}`?"
+                                "Did you mean to use `#{inspect owner_key}`?"
           %_{} ->
             error! query, expr, "field `#{field}` in `#{kind}` is an association in schema #{inspect schema}"
 
diff --git a/lib/ecto/repo/preloader.ex b/lib/ecto/repo/preloader.ex
index 0f071b026e..38a652041d 100644
--- a/lib/ecto/repo/preloader.ex
+++ b/lib/ecto/repo/preloader.ex
@@ -10,7 +10,7 @@ defmodule Ecto.Repo.Preloader do
   Transforms a result set based on query preloads, loading
   the associations onto their parent schema.
   """
-  @spec query([list], Ecto.Repo.t, list, Access.t, fun, {adapter_meta :: map, opts :: Keyword.t}) :: [list]
+  @spec query([list], Ecto.Repo.t(), list, Access.t(), fun, {adapter_meta :: map, opts :: Keyword.t()}) :: [list]
   def query([], _repo_name, _preloads, _take, _fun, _tuplet), do: []
   def query(rows, _repo_name, [], _take, fun, _tuplet), do: Enum.map(rows, fun)
 
@@ -32,8 +32,8 @@ defmodule Ecto.Repo.Preloader do
   @doc """
   Implementation for `Ecto.Repo.preload/2`.
   """
-  @spec preload(structs, atom, atom | list, {adapter_meta :: map, opts :: Keyword.t}) ::
-                structs when structs: [Ecto.Schema.t] | Ecto.Schema.t | nil
+  @spec preload(structs, atom, atom | list, {adapter_meta :: map, opts :: Keyword.t()}) ::
+                structs when structs: [Ecto.Schema.t()] | Ecto.Schema.t() | nil
   def preload(nil, _repo_name, _preloads, _tuplet) do
     nil
   end
@@ -194,46 +194,59 @@ defmodule Ecto.Repo.Preloader do
 
   defp fetch_ids(structs, module, assoc, {_adapter_meta, opts}) do
     %{field: field, owner_key: owner_key, cardinality: card} = assoc
+    owner_keys = List.wrap owner_key
     force? = Keyword.get(opts, :force, false)
 
     Enum.reduce structs, {[], [], []}, fn
       nil, acc ->
         acc
+
       struct, {fetch_ids, loaded_ids, loaded_structs} ->
         assert_struct!(module, struct)
-        %{^owner_key => id, ^field => value} = struct
+        %{^field => value} = struct
+        id = owner_keys |> Enum.map(&Map.fetch!(struct, &1))
         loaded? = Ecto.assoc_loaded?(value) and not force?
 
-        if loaded? and is_nil(id) and not Ecto.Changeset.Relation.empty?(assoc, value) do
-          Logger.warn """
+        if loaded? and Enum.any?(id, &is_nil/1) and
+             not Ecto.Changeset.Relation.empty?(assoc, value) do
+          key_list = "(#{Enum.join(owner_keys, ", ")})"
+
+          Logger.warn("""
           association `#{field}` for `#{inspect(module)}` has a loaded value but \
-          its association key `#{owner_key}` is nil. This usually means one of:
+          its association keys `#{key_list}` are nil. This usually means one of:
 
-            * `#{owner_key}` was not selected in a query
+            * `#{key_list}` were not selected in a query
             * the struct was set with default values for `#{field}` which now you want to override
 
           If this is intentional, set force: true to disable this warning
-          """
+          """)
         end
 
         cond do
           card == :one and loaded? ->
-            {fetch_ids, [id | loaded_ids], [value | loaded_structs]}
+            {fetch_ids, [unwrap_list(id) | loaded_ids], [value | loaded_structs]}
+
           card == :many and loaded? ->
-            {fetch_ids, [{id, length(value)} | loaded_ids], value ++ loaded_structs}
-          is_nil(id) ->
+            {fetch_ids, [{unwrap_list(id), length(value)} | loaded_ids], value ++ loaded_structs}
+
+          Enum.any?(id, &is_nil/1) ->
             {fetch_ids, loaded_ids, loaded_structs}
+
           true ->
-            {[id | fetch_ids], loaded_ids, loaded_structs}
+            {[unwrap_list(id) | fetch_ids], loaded_ids, loaded_structs}
         end
     end
   end
 
-  defp fetch_query(ids, assoc, _repo_name, query, _prefix, related_key, _take, _tuplet) when is_function(query, 1) do
+  defp unwrap_list([id]), do: id
+  defp unwrap_list([_ | _] = ids), do: ids
+
+  defp fetch_query(ids, assoc, _repo_name, query, _prefix, related_key, _take, _tuplet)
+       when is_function(query, 1) do
     # Note we use an explicit sort because we don't want
     # to reorder based on the struct. Only the ID.
     ids
-    |> Enum.uniq
+    |> Enum.uniq()
     |> query.()
     |> fetched_records_to_tuple_ids(assoc, related_key)
     |> Enum.sort(fn {id1, _}, {id2, _} -> id1 <= id2 end)
@@ -242,13 +255,14 @@ defmodule Ecto.Repo.Preloader do
 
   defp fetch_query(ids, %{cardinality: card} = assoc, repo_name, query, prefix, related_key, take, tuplet) do
     query = assoc.__struct__.assoc_query(assoc, query, Enum.uniq(ids))
-    field = related_key_to_field(query, related_key)
+    fields = related_key_to_fields(query, related_key)
 
     # Normalize query
     query = %{Ecto.Query.Planner.ensure_select(query, take || true) | prefix: prefix}
 
     # Add the related key to the query results
-    query = update_in query.select.expr, &{:{}, [], [field, &1]}
+    # TODO: do we have to unwrap field list if it's only one?
+    query = update_in(query.select.expr, &{:{}, [], [unwrap_list(fields), &1]})
 
     # If we are returning many results, we must sort by the key too
     query =
@@ -262,78 +276,95 @@ defmodule Ecto.Repo.Preloader do
 
         {:many, _} ->
           update_in query.order_bys, fn order_bys ->
-            [%Ecto.Query.QueryExpr{expr: preload_order(assoc, query, field), params: [],
+            [%Ecto.Query.QueryExpr{expr: preload_order(assoc, query, fields), params: [],
                                    file: __ENV__.file, line: __ENV__.line}|order_bys]
           end
         {:one, _} ->
           query
       end
 
-    unzip_ids Ecto.Repo.Queryable.all(repo_name, query, tuplet), [], []
+    unzip_ids(Ecto.Repo.Queryable.all(repo_name, query, tuplet), [], [])
   end
 
-  defp fetched_records_to_tuple_ids([], _assoc, _related_key),
-    do: []
+  defp fetched_records_to_tuple_ids([], _assoc, _related_key), do: []
 
-  defp fetched_records_to_tuple_ids([%{} | _] = entries, _assoc, {0, key}),
-    do: Enum.map(entries, &{Map.fetch!(&1, key), &1})
+  defp fetched_records_to_tuple_ids([%{} | _] = entries, _assoc, {0, key_or_keys}) do
+    Enum.map(entries, fn entry ->
+      key = key_or_keys |> List.wrap |> Enum.map(&Map.fetch!(entry, &1)) |> unwrap_list()
+      {key, entry}
+    end)
+  end
 
-  defp fetched_records_to_tuple_ids([{_, %{}} | _] = entries, _assoc, _related_key),
-    do: entries
+  defp fetched_records_to_tuple_ids([{_, %{}} | _] = entries, _assoc, _related_key) do
+    Enum.map(entries, fn {key, value} -> {key, value} end)
+  end
 
   defp fetched_records_to_tuple_ids([entry | _], assoc, _),
-    do: raise """
-    invalid custom preload for `#{assoc.field}` on `#{inspect assoc.owner}`.
-
-    For many_to_many associations, the custom function given to preload should \
-    return a tuple with the associated key as first element and the struct as \
-    second element.
-
-    For example, imagine posts has many to many tags through a posts_tags table. \
-    When preloading the tags, you may write:
-
-        custom_tags = fn post_ids ->
-          Repo.all(
-            from t in Tag,
-                 join: pt in "posts_tags",
-                 where: t.custom and pt.post_id in ^post_ids and pt.tag_id == t.id
-          )
-        end
+    do:
+      raise("""
+      invalid custom preload for `#{assoc.field}` on `#{inspect(assoc.owner)}`.
+
+      For many_to_many associations, the custom function given to preload should \
+      return a tuple with the associated key as first element and the record as \
+      second element.
+
+      For example, imagine posts has many to many tags through a posts_tags table. \
+      When preloading the tags, you may write:
+
+          custom_tags = fn post_ids ->
+            Repo.all(
+              from t in Tag,
+                   join: pt in "posts_tags",
+                   where: t.custom and pt.post_id in ^post_ids and pt.tag_id == t.id
+            )
+          end
 
-        from Post, preload: [tags: ^custom_tags]
+          from Post, preload: [tags: ^custom_tags]
 
-    Unfortunately the query above is not enough because Ecto won't know how to \
-    associate the posts with the tags. In those cases, you need to return a tuple \
-    with the `post_id` as first element and the tag struct as second. The new query \
-    will have a select field as follows:
+      Unfortunately the query above is not enough because Ecto won't know how to \
+      associate the posts with the tags. In those cases, you need to return a tuple \
+      with the `post_id` as first element and the tag record as second. The new query \
+      will have a select field as follows:
 
-        from t in Tag,
-             join: pt in "posts_tags",
-             where: t.custom and pt.post_id in ^post_ids and pt.tag_id == t.id,
-             select: {pt.post_id, t}
+          from t in Tag,
+               join: pt in "posts_tags",
+               where: t.custom and pt.post_id in ^post_ids and pt.tag_id == t.id,
+               select: {pt.post_id, t}
 
-    Expected a tuple with ID and struct, got: #{inspect(entry)}
-    """
+      Expected a tuple with ID and struct, got: #{inspect(entry)}
+      """)
 
-  defp preload_order(assoc, query, related_field) do
-    custom_order_by = Enum.map(assoc.preload_order, fn
-      {direction, field} ->
-        {direction, related_key_to_field(query, {0, field})}
-      field ->
-        {:asc, related_key_to_field(query, {0, field})}
-    end)
+  defp preload_order(assoc, query, related_fields) do
+    custom_order_by =
+      Enum.map(assoc.preload_order, fn
+        {direction, field} ->
+          {direction, related_key_to_fields(query, {0, [field]})}
 
-    [{:asc, related_field} | custom_order_by]
-  end
+        field ->
+          {:asc, related_key_to_fields(query, {0, [field]})}
+      end)
 
-  defp related_key_to_field(query, {pos, key, field_type}) do
-    field_ast = related_key_to_field(query, {pos, key})
+    # fields can be lists and some DB drivers don't support this; flatten out sorting directives to single fields
+    [{:asc, related_fields} | custom_order_by]
+    |> Enum.flat_map(fn {direction, fields} -> Enum.map(fields, &{direction, &1}) end)
+  end
 
-    {:type, [], [field_ast, field_type]}
+  defp related_key_to_fields(query, {pos, key_or_keys}) do
+    key_or_keys
+    |> List.wrap()
+    |> Enum.map(fn key ->
+      {{:., [], [{:&, [], [related_key_pos(query, pos)]}, key]}, [], []}
+    end)
   end
 
-  defp related_key_to_field(query, {pos, key}) do
-    {{:., [], [{:&, [], [related_key_pos(query, pos)]}, key]}, [], []}
+  defp related_key_to_fields(query, {pos, keys, types}) do
+    keys
+    |> List.wrap()
+    |> Enum.zip(List.wrap(types))
+    |> Enum.map(fn {key, field_type} ->
+      field_ast = {{:., [], [{:&, [], [related_key_pos(query, pos)]}, key]}, [], []}
+      {:type, [], [field_ast, field_type]}
+    end)
   end
 
   defp related_key_pos(_query, pos) when pos >= 0, do: pos
@@ -345,31 +376,35 @@ defmodule Ecto.Repo.Preloader do
   defp assert_struct!(mod, %{__struct__: mod}), do: true
   defp assert_struct!(mod, %{__struct__: struct}) do
     raise ArgumentError, "expected a homogeneous list containing the same struct, " <>
-                         "got: #{inspect mod} and #{inspect struct}"
+                         "got: #{inspect(mod)} and #{inspect(struct)}"
   end
 
   defp assoc_map(:one, ids, structs) do
     one_assoc_map(ids, structs, %{})
   end
+
   defp assoc_map(:many, ids, structs) do
     many_assoc_map(ids, structs, %{})
   end
 
-  defp one_assoc_map([id|ids], [struct|structs], map) do
+  defp one_assoc_map([id | ids], [struct | structs], map) do
     one_assoc_map(ids, structs, Map.put(map, id, struct))
   end
+
   defp one_assoc_map([], [], map) do
     map
   end
 
-  defp many_assoc_map([{id, n}|ids], structs, map) do
+  defp many_assoc_map([{id, n} | ids], structs, map) do
     {acc, structs} = split_n(structs, n, [])
     many_assoc_map(ids, structs, Map.put(map, id, acc))
   end
-  defp many_assoc_map([id|ids], [struct|structs], map) do
+
+  defp many_assoc_map([id | ids], [struct | structs], map) do
     {ids, structs, acc} = split_while(ids, structs, id, [struct])
     many_assoc_map(ids, structs, Map.put(map, id, acc))
   end
+
   defp many_assoc_map([], [], map) do
     map
   end
@@ -377,8 +412,9 @@ defmodule Ecto.Repo.Preloader do
   defp split_n(structs, 0, acc), do: {acc, structs}
   defp split_n([struct | structs], n, acc), do: split_n(structs, n - 1, [struct | acc])
 
-  defp split_while([id|ids], [struct|structs], id, acc),
-    do: split_while(ids, structs, id, [struct|acc])
+  defp split_while([id | ids], [struct | structs], id, acc),
+    do: split_while(ids, structs, id, [struct | acc])
+
   defp split_while(ids, structs, _id, acc),
     do: {ids, structs, acc}
 
@@ -390,7 +426,12 @@ defmodule Ecto.Repo.Preloader do
 
   defp load_assoc({:assoc, assoc, ids}, struct) do
     %{field: field, owner_key: owner_key, cardinality: cardinality} = assoc
-    key = Map.fetch!(struct, owner_key)
+
+    key =
+      owner_key
+      |> List.wrap()
+      |> Enum.map(fn owner_key_field -> Map.fetch!(struct, owner_key_field) end)
+      |> unwrap_list()
 
     loaded =
       case ids do
@@ -417,6 +458,7 @@ defmodule Ecto.Repo.Preloader do
 
   defp recur_through(field, {structs, owner}) do
     assoc = owner.__schema__(:association, field)
+
     case assoc.__struct__.preload_info(assoc) do
       {:assoc, %{related: related}, _} ->
         pk_fields =
@@ -437,8 +479,9 @@ defmodule Ecto.Repo.Preloader do
               case set do
                 %{^pk_values => true} ->
                   {fresh, set}
+
                 _ ->
-                  {[child|fresh], Map.put(set, pk_values, true)}
+                  {[child | fresh], Map.put(set, pk_values, true)}
               end
             end)
           end)
@@ -453,7 +496,7 @@ defmodule Ecto.Repo.Preloader do
   defp validate_has_pk_field!([], related, assoc) do
     raise ArgumentError,
           "cannot preload through association `#{assoc.field}` on " <>
-            "`#{inspect assoc.owner}`. Ecto expected the #{inspect related} schema " <>
+            "`#{inspect(assoc.owner)}`. Ecto expected the #{inspect(related)} schema " <>
             "to have at least one primary key field"
   end
 
@@ -467,9 +510,9 @@ defmodule Ecto.Repo.Preloader do
 
         _ ->
           raise ArgumentError,
-               "cannot preload through association `#{assoc.field}` on " <>
-                 "`#{inspect assoc.owner}`. Ecto expected a map/struct with " <>
-                 "the key `#{pk}` but got: #{inspect map}"
+                "cannot preload through association `#{assoc.field}` on " <>
+                  "`#{inspect(assoc.owner)}`. Ecto expected a map/struct with " <>
+                  "the key `#{pk}` but got: #{inspect(map)}"
       end
     end)
   end
@@ -479,8 +522,8 @@ defmodule Ecto.Repo.Preloader do
       [nil | _] ->
         raise ArgumentError,
               "cannot preload through association `#{assoc.field}` on " <>
-                "`#{inspect assoc.owner}` because the primary key `#{hd(pks)}` " <>
-                "is nil for map/struct: #{inspect map}"
+                "`#{inspect(assoc.owner)}` because the primary key `#{hd(pks)}` " <>
+                "is nil for map/struct: #{inspect(map)}"
 
       _ ->
         values
@@ -496,21 +539,25 @@ defmodule Ecto.Repo.Preloader do
   defp normalize_each({atom, {query, list}}, acc, take, original)
        when is_atom(atom) and (is_map(query) or is_function(query, 1)) do
     fields = take(take, atom)
-    [{atom, {fields, query!(query), normalize_each(wrap(list, original), [], fields, original)}}|acc]
+
+    [
+      {atom, {fields, query!(query), normalize_each(wrap(list, original), [], fields, original)}}
+      | acc
+    ]
   end
 
   defp normalize_each({atom, query}, acc, take, _original)
        when is_atom(atom) and (is_map(query) or is_function(query, 1)) do
-    [{atom, {take(take, atom), query!(query), []}}|acc]
+    [{atom, {take(take, atom), query!(query), []}} | acc]
   end
 
   defp normalize_each({atom, list}, acc, take, original) when is_atom(atom) do
     fields = take(take, atom)
-    [{atom, {fields, nil, normalize_each(wrap(list, original), [], fields, original)}}|acc]
+    [{atom, {fields, nil, normalize_each(wrap(list, original), [], fields, original)}} | acc]
   end
 
   defp normalize_each(atom, acc, take, _original) when is_atom(atom) do
-    [{atom, {take(take, atom), nil, []}}|acc]
+    [{atom, {take(take, atom), nil, []}} | acc]
   end
 
   defp normalize_each(other, acc, take, original) do
@@ -529,11 +576,14 @@ defmodule Ecto.Repo.Preloader do
 
   defp wrap(list, _original) when is_list(list),
     do: list
+
   defp wrap(atom, _original) when is_atom(atom),
     do: atom
+
   defp wrap(other, original) do
-    raise ArgumentError, "invalid preload `#{inspect other}` in `#{inspect original}`. " <>
-                         "preload expects an atom, a (nested) keyword or a (nested) list of atoms"
+    raise ArgumentError,
+          "invalid preload `#{inspect(other)}` in `#{inspect(original)}`. " <>
+            "preload expects an atom, a (nested) keyword or a (nested) list of atoms"
   end
 
   ## Expand
@@ -575,11 +625,14 @@ defmodule Ecto.Repo.Preloader do
 
   defp merge_preloads(_preload, {info, _, nil, left}, {info, take, query, right}),
     do: {info, take, query, left ++ right}
+
   defp merge_preloads(_preload, {info, take, query, left}, {info, _, nil, right}),
     do: {info, take, query, left ++ right}
+
   defp merge_preloads(preload, {info, _, left, _}, {info, _, right, _}) do
-    raise ArgumentError, "cannot preload `#{preload}` as it has been supplied more than once " <>
-                         "with different queries: #{inspect left} and #{inspect right}"
+    raise ArgumentError,
+          "cannot preload `#{preload}` as it has been supplied more than once " <>
+            "with different queries: #{inspect(left)} and #{inspect(right)}"
   end
 
   defp association_or_embed!(schema, preload) do
@@ -591,7 +644,7 @@ defmodule Ecto.Repo.Preloader do
     case Atom.to_string(assoc) do
       "Elixir." <> _ ->
         " (if you were trying to pass a schema as a query to preload, " <>
-          "you have to explicitly convert it to a query by doing `from x in #{inspect assoc}` " <>
+          "you have to explicitly convert it to a query by doing `from x in #{inspect(assoc)}` " <>
           "or by calling Ecto.Queryable.to_query/1)"
 
       _ ->
diff --git a/lib/ecto/repo/schema.ex b/lib/ecto/repo/schema.ex
index ec661da406..66fcfed696 100644
--- a/lib/ecto/repo/schema.ex
+++ b/lib/ecto/repo/schema.ex
@@ -905,20 +905,26 @@ defmodule Ecto.Repo.Schema do
   end
 
   defp change_parents(changes, struct, assocs) do
-    Enum.reduce assocs, changes, fn {refl, _}, acc ->
+    Enum.reduce assocs, changes, fn {refl, _}, changes ->
       %{field: field, owner_key: owner_key, related_key: related_key} = refl
       related = Map.get(struct, field)
-      value = related && Map.fetch!(related, related_key)
+      owner_key = List.wrap(owner_key)
+      related_key = List.wrap(related_key)
 
-      case Map.fetch(changes, owner_key) do
-        {:ok, current} when current != value ->
-          raise ArgumentError,
-            "cannot change belongs_to association `#{field}` because there is " <>
-            "already a change setting its foreign key `#{owner_key}` to `#{inspect current}`"
+      Ecto.Association.strict_zip(owner_key, related_key)
+      |> Enum.reduce(changes, fn {owner_key, related_key}, changes ->
+        value = related && Map.fetch!(related, related_key)
 
-        _ ->
-          Map.put(acc, owner_key, value)
-      end
+        case Map.fetch(changes, owner_key) do
+          {:ok, current} when current != value ->
+            raise ArgumentError,
+                  "cannot change belongs_to association `#{field}` because there is " <>
+                  "already a change setting its foreign key `#{owner_key}` to `#{inspect(current)}`"
+
+          _ ->
+            Map.put(changes, owner_key, value)
+        end
+      end)
     end
   end
 
diff --git a/lib/ecto/schema.ex b/lib/ecto/schema.ex
index ef5983f2ba..b56e9b8a8b 100644
--- a/lib/ecto/schema.ex
+++ b/lib/ecto/schema.ex
@@ -771,10 +771,12 @@ defmodule Ecto.Schema do
 
     * `:foreign_key` - Sets the foreign key, this should map to a field on the
       other schema, defaults to the underscored name of the current schema
-      suffixed by `_id`
+      suffixed by `_id`. A list denotes a composite foreign key over multiple
+      columns
 
     * `:references` - Sets the key on the current schema to be used for the
-      association, defaults to the primary key on the schema
+      association, defaults to the primary key on the schema. A list can be
+      provided if the primary key is composite
 
     * `:through` - Allow this association to be defined in terms of existing
       associations. Read the section on `:through` associations for more info
@@ -847,6 +849,20 @@ defmodule Ecto.Schema do
         end
       end
 
+  `has_many` can be defined over a foreign key that spans more than one column.
+
+      defmodule Domain do
+        use Ecto.Schema
+
+        @primary_key false 
+        schema "domains" do
+          field :tld, :string, primary_key: true
+          field :name, :string, primary_key: true
+          has_many :subdomains, Subdomain, foreign_key: [:parent_tld, :parent_name],
+            references: [:tld, :name]
+        end
+      end
+
   ## Filtering associations
 
   It is possible to specify a `:where` option that will filter the records
@@ -1004,10 +1020,12 @@ defmodule Ecto.Schema do
 
     * `:foreign_key` - Sets the foreign key, this should map to a field on the
       other schema, defaults to the underscored name of the current module
-      suffixed by `_id`
+      suffixed by `_id`. A list denotes a composite foreign key over multiple
+      columns
 
     * `:references`  - Sets the key on the current schema to be used for the
-      association, defaults to the primary key on the schema
+      association, defaults to the primary key on the schema. A list can be
+      provided if the primary key is composite
 
     * `:through` - If this association must be defined in terms of existing
       associations. Read the section in `has_many/3` for more information
@@ -1085,16 +1103,21 @@ defmodule Ecto.Schema do
       of the association suffixed by `_id`. For example, `belongs_to :company`
       will define foreign key of `:company_id`. The associated `has_one` or `has_many`
       field in the other schema should also have its `:foreign_key` option set
-      with the same value.
+      with the same value. If the primary key is composite then a list of keys has
+      to be used
 
     * `:references` - Sets the key on the other schema to be used for the
-      association, defaults to: `:id`
+      association, defaults to: `:id`. A list of keys has to be provided if the
+      primary key is composite
 
     * `:define_field` - When false, does not automatically define a `:foreign_key`
       field, implying the user is defining the field manually elsewhere
 
     * `:type` - Sets the type of automatically defined `:foreign_key`.
-      Defaults to: `:integer` and can be set per schema via `@foreign_key_type`
+      Defaults to: `:integer` and can be set per schema via `@foreign_key_type`.
+      A list of types can be used for composite foreign keys. If `:type` is a
+      single atom and the foreign key is composite, the same type is assumed for
+      all key columns
 
     * `:on_replace` - The action taken on associations when the record is
       replaced when casting or manipulating parent changeset. May be
@@ -1145,6 +1168,17 @@ defmodule Ecto.Schema do
         end
       end
 
+  If the references schema has composite primary keys:
+
+      defmodule Comment do
+        use Ecto.Schema
+
+        schema "comments" do
+          belongs_to :publication, Publication, references: [:name, :kind],
+            foreign_key: [:publication_name, :publication_kind], type: [:string, :integer]
+        end
+      end
+
   ## Polymorphic associations
 
   One common use case for belongs to associations is to handle
@@ -1303,7 +1337,9 @@ defmodule Ecto.Schema do
       the join table should reach the current schema and the second
       how the join table should reach the associated schema. In the
       example above, it defaults to: `[post_id: :id, tag_id: :id]`.
-      The keys are inflected from the schema names.
+      The keys are inflected from the schema names. Nested keyword
+      lists have to be used if any of the associated schemas have
+      composite primary keys.
 
     * `:on_delete` - The action taken on associations when the parent record
       is deleted. May be `:nothing` (default) or `:delete_all`.
@@ -1517,6 +1553,25 @@ defmodule Ecto.Schema do
         {:ok, assoc} -> # Assoc was created!
         {:error, changeset} -> # Handle the error
       end
+
+  ## Composite primary key example
+
+  If the current or referenced schema has a composite primary key, `join_keys`
+  has to be a list containing two keyword lists. The first describes the
+  mapping from the join table to the current schema, the second denotes the
+  mapping from the join table to the associated schema.
+
+      defmodule Project do
+        use Ecto.Schema
+
+        @primary_key false
+        schema "word" do
+          field :spelling, :string, primary_key: true
+          field :language, :string, primary_key: true
+          many_to_many :meanings, Meaning, join_through: "words_meanings",
+            join_keys: [[word_spelling: :spelling, word_language: :language], [meaning_id: :id]]
+        end
+      end
   """
   defmacro many_to_many(name, queryable, opts \\ []) do
     queryable = expand_alias(queryable, __CALLER__)
@@ -2042,20 +2097,28 @@ defmodule Ecto.Schema do
 
   @doc false
   def __belongs_to__(mod, name, queryable, opts) do
+    # opts = Keyword.update(opts, :foreign_key, [:"#{name}_id"], &List.wrap/1)
     opts = Keyword.put_new(opts, :foreign_key, :"#{name}_id")
 
-    foreign_key_name = opts[:foreign_key]
-    foreign_key_type = opts[:type] || Module.get_attribute(mod, :foreign_key_type)
-    foreign_key_type = check_field_type!(mod, name, foreign_key_type, opts)
-    check_options!(foreign_key_type, opts, @valid_belongs_to_options, "belongs_to/3")
+    foreign_key_names = opts[:foreign_key] |> List.wrap
+    foreign_key_types = case opts[:type] || Module.get_attribute(mod, :foreign_key_type) do
+      foreign_key_types when is_list(foreign_key_types) ->
+        foreign_key_types
+      foreign_key_type when is_atom(foreign_key_type) ->
+        List.duplicate(foreign_key_type, length(foreign_key_names))
+    end
+    foreign_key_types = Enum.map(foreign_key_types, &check_field_type!(mod, name, &1, opts))
+    Enum.each(foreign_key_types, &check_options!(&1, opts, @valid_belongs_to_options, "belongs_to/3"))
 
-    if foreign_key_name == name do
+    if name in foreign_key_names do
       raise ArgumentError, "foreign_key #{inspect name} must be distinct from corresponding association name"
     end
 
     if Keyword.get(opts, :define_field, true) do
-      Module.put_attribute(mod, :ecto_changeset_fields, {foreign_key_name, foreign_key_type})
-      define_field(mod, foreign_key_name, foreign_key_type, opts)
+      for {foreign_key_name, foreign_key_type} <- Enum.zip(foreign_key_names, foreign_key_types) do
+        Module.put_attribute(mod, :ecto_changeset_fields, {foreign_key_name, foreign_key_type})
+        define_field(mod, foreign_key_name, foreign_key_type, opts)
+      end
     end
 
     struct =
diff --git a/test/ecto/changeset_test.exs b/test/ecto/changeset_test.exs
index 0209049665..38c3ffaf8b 100644
--- a/test/ecto/changeset_test.exs
+++ b/test/ecto/changeset_test.exs
@@ -2228,7 +2228,7 @@ defmodule Ecto.ChangesetTest do
   test "no_assoc_constraint/3 with has_many" do
     changeset = change(%Post{}) |> no_assoc_constraint(:comments)
     assert constraints(changeset) ==
-           [%{type: :foreign_key, field: :comments, constraint: "comments_post_id_fkey", match: :exact,
+           [%{type: :foreign_key, field: :comments, constraint: "comments_post_id_post_token_fkey", match: :exact,
               error_message: "are still associated with this entry", error_type: :no_assoc}]
 
     changeset = change(%Post{}) |> no_assoc_constraint(:comments, name: :whatever, message: "exists")
@@ -2240,7 +2240,7 @@ defmodule Ecto.ChangesetTest do
   test "no_assoc_constraint/3 with has_one" do
     changeset = change(%Post{}) |> no_assoc_constraint(:comment)
     assert constraints(changeset) ==
-           [%{type: :foreign_key, field: :comment, constraint: "comments_post_id_fkey", match: :exact,
+           [%{type: :foreign_key, field: :comment, constraint: "comments_post_id_post_token_fkey", match: :exact,
               error_message: "is still associated with this entry", error_type: :no_assoc}]
 
     changeset = change(%Post{}) |> no_assoc_constraint(:comment, name: :whatever, message: "exists")
diff --git a/test/ecto/query/planner_test.exs b/test/ecto/query/planner_test.exs
index b5fb325702..160d9c8813 100644
--- a/test/ecto/query/planner_test.exs
+++ b/test/ecto/query/planner_test.exs
@@ -105,6 +105,18 @@ defmodule Ecto.Query.PlannerTest do
     def load(data, _, _), do: {:ok, data}
   end
 
+  defmodule CompositePk do
+    use Ecto.Schema
+
+    @primary_key false
+    schema "composites" do
+      field :id_1, :string, primary_key: true
+      field :id_2, :integer, primary_key: true
+
+      many_to_many :posts, Ecto.Query.PlannerTest.Post, join_through: "composites_posts", join_keys: [[composite_id_1: :id_1, composite_id_2: :id_2], [post_id: :id]], join_where: [deleted: true]
+    end
+  end
+
   defmodule Post do
     use Ecto.Schema
 
@@ -131,6 +143,8 @@ defmodule Ecto.Query.PlannerTest do
       many_to_many :crazy_comments, Comment, join_through: CommentPost, where: [text: "crazycomment"]
       many_to_many :crazy_comments_with_list, Comment, join_through: CommentPost, where: [text: {:in, ["crazycomment1", "crazycomment2"]}], join_where: [deleted: true]
       many_to_many :crazy_comments_without_schema, Comment, join_through: "comment_posts", join_where: [deleted: true]
+
+      many_to_many :composites, CompositePk, join_through: "composites_posts", join_keys: [[post_id: :id], [composite_id_1: :id_1, composite_id_2: :id_2]], join_where: [deleted: true]
     end
   end
 
@@ -1105,6 +1119,36 @@ defmodule Ecto.Query.PlannerTest do
     assert dump_params ==  [true, 1]
   end
 
+  test "normalize: many_to_many assoc join with composite keys on association" do
+    {query, cast_params, dump_params, _select} = from(post in Post, join: comment in assoc(post, :composites)) |> normalize_with_params()
+
+    assert inspect(query) =~ "join: c1 in Ecto.Query.PlannerTest.CompositePk, on: c2.composite_id_1 == c1.id_1 and c2.composite_id_2 == c1.id_2 and c2.deleted == ^..."
+    assert cast_params == [true]
+    assert dump_params == [true]
+
+    {query, cast_params, dump_params, _} = Ecto.assoc(%Post{id: 1}, :composites) |> normalize_with_params()
+
+    assert inspect(query) =~ "join: c1 in \"composites_posts\", on: c0.id_1 == c1.composite_id_1 and c0.id_2 == c1.composite_id_2 and c1.deleted == ^..."
+    assert inspect(query) =~ "where: c1.post_id in ^..."
+    assert cast_params ==  [true, 1]
+    assert dump_params ==  [true, 1]
+  end
+
+  test "normalize: many_to_many assoc join with composite keys on owner" do
+    {query, cast_params, dump_params, _} = from(compo in CompositePk, join: post in assoc(compo, :posts)) |> normalize_with_params()
+
+    assert inspect(query) =~ "join: p1 in Ecto.Query.PlannerTest.Post, on: c2.post_id == p1.id and c2.deleted == ^..."
+    assert cast_params == [true]
+    assert dump_params == [true]
+
+    {query, cast_params, dump_params, _} = Ecto.assoc(%Post{id: 1}, :composites) |> normalize_with_params()
+
+    assert inspect(query) =~ "join: c1 in \"composites_posts\", on: c0.id_1 == c1.composite_id_1 and c0.id_2 == c1.composite_id_2 and c1.deleted == ^..."
+    assert inspect(query) =~ "where: c1.post_id in ^..."
+    assert cast_params ==  [true, 1]
+    assert dump_params ==  [true, 1]
+  end
+
   test "normalize: dumps in query expressions" do
     assert_raise Ecto.QueryError, ~r"cannot be dumped", fn ->
       normalize(from p in Post, where: p.posted == "2014-04-17 00:00:00")
@@ -1135,7 +1179,7 @@ defmodule Ecto.Query.PlannerTest do
     end
 
     message =
-      ~r"field `crazy_post_with_list` in `select` is an association in schema Ecto.Query.PlannerTest.Comment. Did you mean to use `crazy_post_id`"
+      ~r"field `crazy_post_with_list` in `select` is an association in schema Ecto.Query.PlannerTest.Comment. Did you mean to use [`crazy_post_id`]"
     assert_raise Ecto.QueryError, message, fn ->
       query = from(Comment, []) |> select([c], c.crazy_post_with_list)
       normalize(query)
diff --git a/test/ecto/repo/belongs_to_test.exs b/test/ecto/repo/belongs_to_test.exs
index fceaac9e77..5cad3f631f 100644
--- a/test/ecto/repo/belongs_to_test.exs
+++ b/test/ecto/repo/belongs_to_test.exs
@@ -24,6 +24,20 @@ defmodule Ecto.Repo.BelongsToTest do
     end
   end
 
+  defmodule MyCompositeAssoc do
+    use Ecto.Schema
+
+    @primary_key false
+    schema "my_composite_assoc" do
+      field :id_1, :id,  primary_key: true
+      field :id_2, :string, primary_key: true
+      field :name, :string
+      has_one :my_schema, MySchema,
+        foreign_key: [:composite_id_1, :composite_id_2], references: [:id_1, :id_2]
+      timestamps()
+    end
+  end
+
   defmodule MySchema do
     use Ecto.Schema
 
@@ -32,6 +46,8 @@ defmodule Ecto.Repo.BelongsToTest do
       field :y, :binary
       belongs_to :assoc, MyAssoc, on_replace: :delete
       belongs_to :nilify_assoc, MyAssoc, on_replace: :nilify
+      belongs_to :composite_assoc, MyCompositeAssoc,
+        foreign_key: [:composite_id_1, :composite_id_2], references: [:id_1, :id_2], type: [:id, :string]
     end
   end
 
@@ -50,6 +66,22 @@ defmodule Ecto.Repo.BelongsToTest do
     assert assoc.inserted_at
   end
 
+  test "handles assocs with composite keys on insert" do
+    sample = %MyCompositeAssoc{id_1: 123, id_2: "xyz"}
+
+    changeset =
+      %MySchema{}
+      |> Ecto.Changeset.change
+      |> Ecto.Changeset.put_assoc(:composite_assoc, sample)
+    schema = TestRepo.insert!(changeset)
+    assoc = schema.composite_assoc
+    assert assoc.id_1 == 123
+    assert assoc.id_2 == "xyz"
+    assert assoc.id_1 == schema.composite_id_1
+    assert assoc.id_2 == schema.composite_id_2
+    assert assoc.inserted_at
+  end
+
   test "handles assocs on insert preserving parent schema prefix" do
     sample = %MyAssoc{x: "xyz"}
 
@@ -106,6 +138,32 @@ defmodule Ecto.Repo.BelongsToTest do
     end
   end
 
+  test "checks dual changes to composite keys on insert" do
+    # values are the same
+    changeset =
+      %MySchema{}
+      |> Ecto.Changeset.change(composite_id_1: 13)
+      |> Ecto.Changeset.put_assoc(:composite_assoc, %MyCompositeAssoc{id_1: 13, id_2: "xyz"})
+    TestRepo.insert!(changeset)
+
+    # values are different
+    changeset =
+      %MySchema{}
+      |> Ecto.Changeset.change(composite_id_1: 13)
+      |> Ecto.Changeset.put_assoc(:composite_assoc, %MyCompositeAssoc{id_2: "xyz"})
+    assert_raise ArgumentError, ~r"there is already a change setting its foreign key", fn ->
+      TestRepo.insert!(changeset)
+    end
+
+    changeset =
+      %MySchema{}
+      |> Ecto.Changeset.change(composite_id_2: "abc")
+      |> Ecto.Changeset.put_assoc(:composite_assoc, %MyCompositeAssoc{id_1: 13})
+    assert_raise ArgumentError, ~r"there is already a change setting its foreign key", fn ->
+      TestRepo.insert!(changeset)
+    end
+  end
+
   test "returns untouched changeset on invalid children on insert" do
     assoc = %MyAssoc{x: "xyz"}
     assoc_changeset = %{Ecto.Changeset.change(assoc) | valid?: false}
@@ -242,7 +300,23 @@ defmodule Ecto.Repo.BelongsToTest do
     assert assoc.updated_at
   end
 
-    test "inserting assocs on update preserving parent schema prefix" do
+  test "inserting assocs with composite keys on update" do
+    sample = %MyCompositeAssoc{id_1: 123, id_2: "xyz"}
+
+    changeset =
+      %MySchema{id: 1}
+      |> Ecto.Changeset.change
+      |> Ecto.Changeset.put_assoc(:composite_assoc, sample)
+    schema = TestRepo.update!(changeset)
+    assoc = schema.composite_assoc
+    assert assoc.id_1 == 123
+    assert assoc.id_2 == "xyz"
+    assert assoc.id_1 == schema.composite_id_1
+    assert assoc.id_2 == schema.composite_id_2
+    assert assoc.inserted_at
+  end
+
+  test "inserting assocs on update preserving parent schema prefix" do
     sample = %MyAssoc{x: "xyz"}
 
     changeset =
@@ -349,6 +423,26 @@ defmodule Ecto.Repo.BelongsToTest do
     refute_received {:delete, _} # Same assoc should not emit delete
   end
 
+  test "changing assocs with composite keys on update" do
+    sample = %MyCompositeAssoc{id_1: 13, id_2: "xyz", name: "old name"}
+    sample = put_meta sample, state: :loaded
+
+    # Changing the assoc
+    sample_changeset = Ecto.Changeset.change(sample, name: "new name")
+    changeset =
+      %MySchema{id: 1, composite_assoc: sample}
+      |> Ecto.Changeset.change
+      |> Ecto.Changeset.put_assoc(:composite_assoc, sample_changeset)
+    schema = TestRepo.update!(changeset)
+    assoc = schema.composite_assoc
+    assert assoc.id_1 == 13
+    assert assoc.id_2 == "xyz"
+    assert assoc.name == "new name"
+    refute assoc.inserted_at
+    assert assoc.updated_at
+    refute_received {:delete, _} # Same assoc should not emit delete
+  end
+
   test "removing assocs on update raises if there is no id" do
     assoc = %MyAssoc{x: "xyz"} |> Ecto.put_meta(state: :loaded)
 
diff --git a/test/ecto/repo/has_assoc_test.exs b/test/ecto/repo/has_assoc_test.exs
index b2af1efe33..a8c058a800 100644
--- a/test/ecto/repo/has_assoc_test.exs
+++ b/test/ecto/repo/has_assoc_test.exs
@@ -570,4 +570,83 @@ defmodule Ecto.Repo.HasAssocTest do
     loaded = put_in schema.__meta__.state, :loaded
     TestRepo.insert!(loaded)
   end
+
+  defmodule MyCompositeAssoc do
+    use Ecto.Schema
+
+    schema "my_composite_assoc" do
+      field :name, :binary
+      belongs_to :composite_schema, MyCompositeSchema,
+        foreign_key: [:composite_x, :composite_y], references: [:x, :y], type: [:id, :string]
+      timestamps()
+    end
+  end
+
+  defmodule MyCompositeSchema do
+    use Ecto.Schema
+
+    @primary_key false
+    schema "my_composite_schema" do
+      field :x, :id, primary_key: true
+      field :y, :string, primary_key: true
+      has_one :assoc, MyCompositeAssoc, foreign_key: [:composite_x, :composite_y], references: [:x, :y]
+      has_many :assocs, MyCompositeAssoc, foreign_key: [:composite_x, :composite_y], references: [:x, :y]
+    end
+  end
+
+  test "handles assocs with composite keys on insert" do
+    sample = %MyCompositeAssoc{name: "xyz"}
+
+    changeset =
+      %MyCompositeSchema{}
+      |> Ecto.Changeset.change
+      |> Ecto.Changeset.put_assoc(:assoc, sample)
+    schema = TestRepo.insert!(changeset)
+    assoc = schema.assoc
+    assert assoc.id
+    assert assoc.name == "xyz"
+    assert assoc.composite_x == schema.x
+    assert assoc.composite_y == schema.y
+    assert assoc.inserted_at
+
+    changeset =
+      %MyCompositeSchema{}
+      |> Ecto.Changeset.change
+      |> Ecto.Changeset.put_assoc(:assocs, [sample])
+    schema = TestRepo.insert!(changeset)
+    [assoc] = schema.assocs
+    assert assoc.id
+    assert assoc.name == "xyz"
+    assert assoc.composite_x == schema.x
+    assert assoc.composite_y == schema.y
+    assert assoc.inserted_at
+  end
+
+  test "handles assocs with composite keys on update" do
+    sample = %MyCompositeAssoc{name: "xyz"}
+
+    changeset =
+      %MyCompositeSchema{x: 3, y: "a"}
+      |> Ecto.Changeset.change
+      |> Ecto.Changeset.put_assoc(:assoc, sample)
+    schema = TestRepo.update!(changeset)
+    assoc = schema.assoc
+    assert assoc.id
+    assert assoc.name == "xyz"
+    assert assoc.composite_x == 3
+    assert assoc.composite_y == "a"
+    assert assoc.updated_at
+
+    changeset =
+      %MyCompositeSchema{x: 4, y: "b"}
+      |> Ecto.Changeset.change
+      |> Ecto.Changeset.put_assoc(:assocs, [sample])
+    schema = TestRepo.update!(changeset)
+    [assoc] = schema.assocs
+    assert assoc.id
+    assert assoc.name == "xyz"
+    assert assoc.composite_x == 4
+    assert assoc.composite_y == "b" 
+    assert assoc.updated_at
+  end
 end
diff --git a/test/ecto/repo/many_to_many_test.exs b/test/ecto/repo/many_to_many_test.exs
index 0bf30f1f9d..8cf3c7d81e 100644
--- a/test/ecto/repo/many_to_many_test.exs
+++ b/test/ecto/repo/many_to_many_test.exs
@@ -46,6 +46,33 @@ defmodule Ecto.Repo.ManyToManyTest do
     end
   end
 
+  defmodule MyCompositeAssoc do
+    use Ecto.Schema
+
+    @primary_key false
+    schema "my_composite_assoc" do
+      field :id_1, :id,  primary_key: true
+      field :id_2, :string, primary_key: true
+      field :name, :string
+      many_to_many :my_schemas, Ecto.Repo.ManyToManyTest.MySchema,
+        join_through: "schemas_composite_assocs",
+        join_keys: [[composite_id_1: :id_1, composite_id_2: :id_2],[my_schema_id: :id]]
+      timestamps()
+    end
+  end
+
+  defmodule MySchemaCompositeAssoc do
+    use Ecto.Schema
+
+    schema "schemas_composite_assocs" do
+      field :public, :boolean, default: false
+      belongs_to :my_schema, MySchema
+      belongs_to :my_assoc, MyCompositeAssoc,
+        foreign_key: [:composite_id_1, :composite_id_2], references: [:id_1, :id_2], type: [:id, :string]
+      timestamps()
+    end
+  end
+
   defmodule MySchema do
     use Ecto.Schema
 
@@ -56,7 +83,15 @@ defmodule Ecto.Repo.ManyToManyTest do
       many_to_many :where_assocs, MyAssoc, join_through: "schemas_assocs", join_where: [public: true], on_replace: :delete
       many_to_many :schema_assocs, MyAssoc, join_through: MySchemaAssoc, join_defaults: [public: true]
       many_to_many :schema_prefix_assocs, MyAssoc, join_through: MySchemaPrefixAssoc, join_defaults: [public: true]
+      many_to_many :schema_key_assocs, MyAssoc, join_through: MySchemaAssoc, join_keys: [my_schema_id: :id, my_assoc_id: :id]
       many_to_many :mfa_schema_assocs, MyAssoc, join_through: MySchemaAssoc, join_defaults: {__MODULE__, :send_to_self, [:extra]}
+      many_to_many :composite_assocs, MyCompositeAssoc,
+        join_through: "schemas_composite_assocs",
+        join_keys: [[my_schema_id: :id], [composite_id_1: :id_1, composite_id_2: :id_2]]
+      many_to_many :schema_composite_assocs, MyCompositeAssoc,
+        join_through: MySchemaCompositeAssoc,
+        join_keys: [[my_schema_id: :id], [composite_id_1: :id_1, composite_id_2: :id_2]],
+        join_defaults: [public: true]
     end
 
     def send_to_self(struct, owner, extra) do
@@ -133,6 +168,28 @@ defmodule Ecto.Repo.ManyToManyTest do
     assert join.fields[:public]
   end
 
+  test "handles assocs on insert with schema and join keys" do
+    sample = %MyAssoc{x: "xyz"}
+
+    changeset =
+      %MySchema{}
+      |> Ecto.Changeset.change
+      |> Ecto.Changeset.put_assoc(:schema_key_assocs, [sample])
+
+    schema = TestRepo.insert!(changeset)
+    [assoc] = schema.schema_key_assocs
+    assert assoc.id
+    assert assoc.x == "xyz"
+    assert assoc.inserted_at
+    assert_received {:insert, _child}
+    assert_received {:insert, _parent}
+    assert_received {:insert, join}
+
+    # Available from defaults
+    assert join.fields[:my_schema_id] == schema.id
+    assert join.fields[:my_assoc_id] == assoc.id
+  end
+
   test "handles assocs on insert with schema and MFA defaults" do
     sample = %MyAssoc{x: "xyz"}
 
@@ -188,6 +245,67 @@ defmodule Ecto.Repo.ManyToManyTest do
     assert_received {:insert, %{source: "schemas_prefix_assocs", prefix: "schema_assoc_prefix"}}
   end
 
+  test "handles assocs with composite keys on insert (parent has normal keys)" do
+    sample = %MyCompositeAssoc{id_1: 1, id_2: "a", name: "xyz"}
+
+    changeset =
+      %MySchema{x: "abc"}
+      |> Ecto.Changeset.change
+      |> Ecto.Changeset.put_assoc(:composite_assocs, [sample])
+    schema = TestRepo.insert!(changeset)
+    [assoc] = schema.composite_assocs
+    assert assoc.id_1 == 1
+    assert assoc.id_2 == "a"
+    assert assoc.name == "xyz"
+    assert assoc.inserted_at
+    assert_received {:insert, _}
+    assert_received {:insert_all, %{source: "schemas_composite_assocs"}, [
+      [composite_id_1: 1, composite_id_2: "a", my_schema_id: 1]
+    ]}
+  end
+
+  test "handles assocs with composite keys on insert (parent has composite keys)" do
+    sample = %MySchema{x: "abc"}
+
+    changeset =
+      %MyCompositeAssoc{id_1: 1, id_2: "a", name: "xyz"}
+      |> Ecto.Changeset.change
+      |> Ecto.Changeset.put_assoc(:my_schemas, [sample])
+    schema = TestRepo.insert!(changeset)
+    [assoc] = schema.my_schemas
+    assert assoc.id
+    assert assoc.x == "abc"
+    assert_received {:insert, _}
+    assert_received {:insert_all, %{source: "schemas_composite_assocs"}, [
+      [composite_id_1: 1, composite_id_2: "a", my_schema_id: 1]
+    ]}
+  end
+
+  test "handles assocs with composite keys and schema on insert" do
+    sample = %MyCompositeAssoc{id_1: 1, id_2: "a", name: "xyz"}
+
+    changeset =
+      %MySchema{x: "abc"}
+      |> Ecto.Changeset.change
+      |> Ecto.Changeset.put_assoc(:schema_composite_assocs, [sample])
+    schema = TestRepo.insert!(changeset)
+    [assoc] = schema.schema_composite_assocs
+    assert assoc.id_1 == 1
+    assert assoc.id_2 == "a"
+    assert assoc.name == "xyz"
+    assert assoc.inserted_at
+
+    assert_received {:insert, _child}
+    assert_received {:insert, _parent}
+    assert_received {:insert, join}
+
+    # Available from defaults
+    assert join.fields[:my_schema_id] == schema.id
+    assert join.fields[:composite_id_1] == assoc.id_1
+    assert join.fields[:composite_id_2] == assoc.id_2
+    assert join.fields[:public]
+  end
+
   test "handles assocs from struct on insert" do
     schema = TestRepo.insert!(%MySchema{assocs: [%MyAssoc{x: "xyz"}]})
     [assoc] = schema.assocs
diff --git a/test/ecto/schema_test.exs b/test/ecto/schema_test.exs
index a5f3eb85ea..a54a59bd37 100644
--- a/test/ecto/schema_test.exs
+++ b/test/ecto/schema_test.exs
@@ -751,6 +751,24 @@ defmodule Ecto.SchemaTest do
       has_one :author, User, references: :pk, foreign_key: :fk
       belongs_to :permalink1, Permalink, references: :pk, foreign_key: :fk
       belongs_to :permalink2, Permalink, references: :pk, type: :string
+      belongs_to :publication1, Publication, references: [:id_1, :id_2],
+        foreign_key: [:publication1_id_1, :publication1_id_2], type: [:integer, :string]
+      belongs_to :publication2, Publication, references: [:id_1, :id_2],
+        foreign_key: [:publication2_id_1, :publication2_id_2], type: :string
+    end
+  end
+
+  defmodule Publication do
+    use Ecto.Schema
+
+    @primary_key false
+    schema "publication" do
+      field :id_1, :integer, primary_key: true
+      field :id_2, :string, primary_key: true
+      has_many :custom_assoc_schemas, CustomAssocSchema, references: [:id_1, :id_2],
+        foreign_key: [:publication_id_1, :publication_id_2]
+      has_one :custom_assoc_schema, CustomAssocSchema, references: [:id_1, :id_2],
+        foreign_key: [:pub_id_1, :pub_id_2]
     end
   end
 
@@ -758,12 +776,20 @@ defmodule Ecto.SchemaTest do
     refl = CustomAssocSchema.__schema__(:association, :posts)
     assert :pk == refl.owner_key
     assert :fk == refl.related_key
+
+    refl = Publication.__schema__(:association, :custom_assoc_schemas)
+    assert [:id_1, :id_2] == refl.owner_key
+    assert [:publication_id_1, :publication_id_2] == refl.related_key
   end
 
   test "has_one options" do
     refl = CustomAssocSchema.__schema__(:association, :author)
     assert :pk == refl.owner_key
     assert :fk == refl.related_key
+
+    refl = Publication.__schema__(:association, :custom_assoc_schema)
+    assert [:id_1, :id_2] == refl.owner_key
+    assert [:pub_id_1, :pub_id_2] == refl.related_key
   end
 
   test "belongs_to options" do
@@ -777,6 +803,17 @@ defmodule Ecto.SchemaTest do
 
     assert CustomAssocSchema.__schema__(:type, :fk) == :string
     assert CustomAssocSchema.__schema__(:type, :permalink2_id) == :string
+
+    refl = CustomAssocSchema.__schema__(:association, :publication1)
+    assert [:publication1_id_1, :publication1_id_2] == refl.owner_key
+    assert [:id_1, :id_2] == refl.related_key
+
+    assert CustomAssocSchema.__schema__(:type, :publication1_id_1) == :integer
+    assert CustomAssocSchema.__schema__(:type, :publication1_id_2) == :string
+
+    # if foreign key type is an atom, use that same type for all foreign key fields
+    assert CustomAssocSchema.__schema__(:type, :publication2_id_1) == :string
+    assert CustomAssocSchema.__schema__(:type, :publication2_id_2) == :string
   end
 
   test "has_* validates option" do
@@ -834,6 +871,20 @@ defmodule Ecto.SchemaTest do
     end
   end
 
+  test "has_* references option has to be set when no primary keys" do
+    message = ~r"need to set :references option for association :posts when schema has no primary key"
+    assert_raise ArgumentError, message, fn ->
+      defmodule CompositePkAssocNoConfig do
+        use Ecto.Schema
+
+        @primary_key false
+        schema "assoc" do
+          has_many :posts, Post
+        end
+      end
+    end
+  end
+
   test "has_* expects a queryable" do
     message = ~r"association :posts queryable must be a schema or a {source, schema}. got: 123"
     assert_raise ArgumentError, message, fn ->