diff --git a/Earthfile b/Earthfile
index 45757abce2..04f146e98c 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 e2089f4f30..bd74a00222 100644
--- a/integration_test/cases/assoc.exs
+++ b/integration_test/cases/assoc.exs
@@ -10,6 +10,7 @@ defmodule Ecto.Integration.AssocTest do
   alias Ecto.Integration.PostUser
   alias Ecto.Integration.Comment
   alias Ecto.Integration.Permalink
+  alias Ecto.Integration.CompositePk
 
   test "has_many assoc" do
     p1 = TestRepo.insert!(%Post{title: "1"})
@@ -750,6 +751,12 @@ 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/repo.exs b/integration_test/cases/repo.exs
index 005bbde5c5..cfb3f7085c 100644
--- a/integration_test/cases/repo.exs
+++ b/integration_test/cases/repo.exs
@@ -152,6 +152,24 @@ defmodule Ecto.Integration.RepoTest do
     assert TestRepo.all(PostUserCompositePk) == []
   end
 
+  @tag :composite_pk
+  # TODO this needs a better name
+  test "insert, update and delete with associated composite pk #2" 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 b97d63a0f6..aa94e0dda2 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]
@@ -291,6 +293,7 @@ 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]
   end
   def changeset(schema, params) do
     cast(schema, params, ~w(a b name)a)
diff --git a/lib/ecto.ex b/lib/ecto.ex
index b16e2deceb..29112c2b4d 100644
--- a/lib/ecto.ex
+++ b/lib/ecto.ex
@@ -510,10 +510,15 @@ 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 ->
+        owner_key
+        # TODO remove List.wrap once all assocs use lists
+        |> List.wrap
+        |> Enum.map(&Map.fetch!(struct, &1))
+      end)
+      |> Enum.uniq
 
     case assocs do
       [] ->
diff --git a/lib/ecto/association.ex b/lib/ecto/association.ex
index 3169be3f9b..673409afbe 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,15 @@ 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))
+      # TODO remove this once all relations store keys in lists
+      in_keys = List.wrap(curr_rel.in_key)
+      out_keys = List.wrap(prev_rel.out_key)
+      next = query
+        # join on the first field of the foreign key
+        |> join(:inner, [{src, counter}], dest in ^related_queryable, on: field(src, ^hd(out_keys)) == field(dest, ^hd(in_keys)))
+        # add the rest of the foreign key fields, if any
+        |> composite_joins_query(counter, counter + 1, tl(out_keys), tl(in_keys))
+        # consider where clauses on assocs
         |> combine_joins_query(curr_rel.where, counter + 1)
 
       {curr_rel, next, counter + 1, curr_rel.out_key}
@@ -320,6 +328,16 @@ defmodule Ecto.Association do
     end)
   end
 
+  # TODO docs
+  def composite_joins_query(query, _binding_src, _binding_dst, [], []) do
+    query
+  end
+  def composite_joins_query(query, binding_src, binding_dst, [src_key | src_keys], [dst_key | dst_keys]) do
+    # TODO
+    [query, binding_src, binding_dst, [src_key | src_keys], [dst_key | dst_keys]] |> IO.inspect(label: :composite_joins_query)
+    query
+  end
+
   @doc """
   Add the default assoc query where clauses to a join.
 
@@ -335,6 +353,16 @@ defmodule Ecto.Association do
     %{query | joins: joins ++ [%{join_expr | on: %{join_on | expr: expr, params: params}}]}
   end
 
+  # TODO docs
+  def composite_assoc_query(query, _binding_src, [], []) do
+    query
+  end
+  def composite_assoc_query(query, binding_dst, [dst_key | dst_keys], [value | values]) do
+    # TODO
+    [query, binding_dst, [dst_key | dst_keys], [value | values]] |> IO.inspect(label: :composite_assoc_query)
+    query
+  end
+
   @doc """
   Add the default assoc query where clauses a provided query.
   """
@@ -632,6 +660,10 @@ 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
+    Enum.filter related_key, &is_nil(queryable.__schema__(:type, &1))
+  end
 end
 
 defmodule Ecto.Association.Has do
@@ -644,8 +676,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 list of columns that form the key on the `owner` schema used for the association
+    * `related_key` - The 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
@@ -673,8 +705,8 @@ 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
@@ -686,14 +718,17 @@ defmodule Ecto.Association.Has do
     cardinality = Keyword.fetch!(opts, :cardinality)
     related = Ecto.Association.related_from_query(queryable, name)
 
-    ref =
+    refs =
       module
       |> Module.get_attribute(:primary_key)
       |> get_ref(opts[:references], name)
+      |> List.wrap()
 
-    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 <- refs 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
@@ -725,13 +760,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 -> Enum.map(refs, &Ecto.Association.association_key(module, &1))
+      key when is_atom(key) -> [key]
+      keys when is_list(keys) -> 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),
+      owner_key: refs,
+      related_key: foreign_key,
       queryable: queryable,
       on_delete: on_delete,
       on_replace: on_replace,
@@ -756,19 +797,23 @@ 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))
+    # TODO find out how to handle a dynamic list of fields here
+    from(o in owner, join: q in ^queryable, on: field(q, ^hd(related_key)) == field(o, ^hd(owner_key)))
+    |> Ecto.Association.composite_joins_query(0, 1, tl(related_key), tl(owner_key))
     |> 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)
+    from(x in (query || queryable), where: field(x, ^hd(related_key)) == ^hd(value))
+    |> Ecto.Association.composite_assoc_query(0, tl(related_key), tl(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), where: field(x, ^hd(related_key)) in ^Enum.map(values, &hd/1))
+    |> Ecto.Association.composite_assoc_query(0, tl(related_key), Enum.map(values, &tl/1))
     |> Ecto.Association.combine_assoc_query(assoc.where)
   end
 
@@ -807,16 +852,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
 
@@ -825,11 +875,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
@@ -982,8 +1042,8 @@ 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}`"}
       true ->
         :ok
     end
@@ -991,7 +1051,7 @@ defmodule Ecto.Association.BelongsTo do
 
   @impl true
   def struct(module, name, opts) do
-    ref = if ref = opts[:references], do: ref, else: :id
+    refs = if ref = opts[:references], do: List.wrap(ref), else: [:id]
     queryable = Keyword.fetch!(opts, :queryable)
     related = Ecto.Association.related_from_query(queryable, name)
     on_replace = Keyword.get(opts, :on_replace, :raise)
@@ -1013,8 +1073,8 @@ defmodule Ecto.Association.BelongsTo do
       field: name,
       owner: module,
       related: related,
-      owner_key: Keyword.fetch!(opts, :foreign_key),
-      related_key: ref,
+      owner_key: List.wrap(Keyword.fetch!(opts, :foreign_key)),
+      related_key: refs,
       queryable: queryable,
       on_replace: on_replace,
       defaults: defaults,
@@ -1031,19 +1091,22 @@ 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, join: q in ^queryable, on: field(q, ^hd(related_key)) == field(o, ^hd(owner_key)))
+    |> Ecto.Association.composite_joins_query(0, 1, tl(related_key), tl(owner_key))
     |> 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)
+    from(x in (query || queryable), where: field(x, ^hd(related_key)) == ^hd(value))
+    |> Ecto.Association.composite_assoc_query(0, tl(related_key), tl(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), where: field(x, ^hd(related_key)) in ^Enum.map(values, &hd/1))
+    |> Ecto.Association.composite_assoc_query(0, tl(related_key), Enum.map(values, &tl/1))
     |> Ecto.Association.combine_assoc_query(assoc.where)
   end
 
@@ -1264,11 +1327,12 @@ defmodule Ecto.Association.ManyToMany do
 
     owner_key_type = owner.__schema__(:type, owner_key)
 
+    # TODO fix the hd(values)
     # 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.
     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})
+      where: field(j, ^join_owner_key) in type(^hd(values), {:in, ^owner_key_type})
     )
     |> Ecto.Association.combine_assoc_query(assoc.where)
     |> Ecto.Association.combine_joins_query(assoc.join_where, 1)
diff --git a/lib/ecto/changeset.ex b/lib/ecto/changeset.ex
index fcf32336c8..dc9e7a905c 100644
--- a/lib/ecto/changeset.ex
+++ b/lib/ecto/changeset.ex
@@ -2794,7 +2794,7 @@ defmodule Ecto.Changeset do
     constraint = opts[:name] ||
       case get_assoc(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}"
@@ -2845,7 +2845,7 @@ defmodule Ecto.Changeset do
       case get_assoc(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,
@@ -3062,6 +3062,12 @@ defmodule Ecto.Changeset do
     |> merge_keyword_keys(msg_func, changeset)
     |> merge_related_keys(changes, types, msg_func, &traverse_validations/2)
   end
+
+  defp atom_concat(atoms) do
+    atoms
+    |> Enum.map(&to_string/1)
+    |> Enum.join("_")
+  end
 end
 
 defimpl Inspect, for: Ecto.Changeset do
diff --git a/lib/ecto/repo/preloader.ex b/lib/ecto/repo/preloader.ex
index b9713ae39f..210dd86905 100644
--- a/lib/ecto/repo/preloader.ex
+++ b/lib/ecto/repo/preloader.ex
@@ -170,10 +170,15 @@ defmodule Ecto.Repo.Preloader do
         acc
       struct, {fetch_ids, loaded_ids, loaded_structs} ->
         assert_struct!(module, struct)
-        %{^owner_key => id, ^field => value} = struct
+        %{^field => value} = struct
+        # TODO remove List.wrap
+        ids = owner_key |> List.wrap |> Enum.map(fn key -> 
+          %{^key => id} = struct
+          id
+        end)
         loaded? = Ecto.assoc_loaded?(value) and not force?
 
-        if loaded? and is_nil(id) and not Ecto.Changeset.Relation.empty?(assoc, value) do
+        if loaded? and Enum.any?(ids, &is_nil/1) and not Ecto.Changeset.Relation.empty?(assoc, value) do
           Logger.warn """
           association `#{field}` for `#{inspect(module)}` has a loaded value but \
           its association key `#{owner_key}` is nil. This usually means one of:
@@ -187,13 +192,13 @@ defmodule Ecto.Repo.Preloader do
 
         cond do
           card == :one and loaded? ->
-            {fetch_ids, [id | loaded_ids], [value | loaded_structs]}
+            {fetch_ids, [ids | loaded_ids], [value | loaded_structs]}
           card == :many and loaded? ->
-            {fetch_ids, [{id, length(value)} | loaded_ids], value ++ loaded_structs}
-          is_nil(id) ->
+            {fetch_ids, [{ids, length(value)} | loaded_ids], value ++ loaded_structs}
+          Enum.any? ids, &is_nil/1 ->
             {fetch_ids, loaded_ids, loaded_structs}
           true ->
-            {[id | fetch_ids], loaded_ids, loaded_structs}
+            {[ids | fetch_ids], loaded_ids, loaded_structs}
         end
     end
   end
@@ -211,20 +216,20 @@ defmodule Ecto.Repo.Preloader do
 
   defp fetch_query(ids, %{cardinality: card} = assoc, repo_name, query, prefix, related_key, take, opts) 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]}
+    query = update_in query.select.expr, &{:{}, [], [fields, &1]}
 
     # If we are returning many results, we must sort by the key too
     query =
       case card 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 ->
@@ -288,6 +293,13 @@ defmodule Ecto.Repo.Preloader do
     [{:asc, related_field} | custom_order_by]
   end
 
+  defp related_key_to_fields(query, {pos, keys}) do
+    Enum.map keys, fn key ->
+      {{:., [], [{:&, [], [related_key_pos(query, pos)]}, key]}, [], []}
+    end
+  end
+
+  # TODO deprecated
   defp related_key_to_field(query, {pos, key, field_type}) do
     field_ast = related_key_to_field(query, {pos, key})
 
@@ -352,16 +364,18 @@ 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)
+    Enum.reduce owner_key, struct, fn owner_key_field, struct ->
+      key = Map.fetch!(struct, owner_key_field)
 
-    loaded =
-      case ids do
-        %{^key => value} -> value
-        _ when cardinality == :many -> []
-        _ -> nil
-      end
+      loaded =
+        case ids do
+          %{^key => value} -> value
+          _ when cardinality == :many -> []
+          _ -> nil
+        end
 
-    Map.put(struct, field, loaded)
+      Map.put(struct, field, loaded)
+    end
   end
 
   defp load_through({:through, assoc, throughs}, struct) do
diff --git a/lib/ecto/repo/schema.ex b/lib/ecto/repo/schema.ex
index ab05f0b0e1..91fe5201b4 100644
--- a/lib/ecto/repo/schema.ex
+++ b/lib/ecto/repo/schema.ex
@@ -13,8 +13,15 @@ defmodule Ecto.Repo.Schema do
   Implementation for `Ecto.Repo.insert_all/3`.
   """
   def insert_all(repo, name, schema, rows, opts) when is_atom(schema) do
-    do_insert_all(repo, name, schema, schema.__schema__(:prefix),
-                  schema.__schema__(:source), rows, opts)
+    do_insert_all(
+      repo,
+      name,
+      schema,
+      schema.__schema__(:prefix),
+      schema.__schema__(:source),
+      rows,
+      opts
+    )
   end
 
   def insert_all(repo, name, table, rows, opts) when is_binary(table) do
@@ -45,7 +52,16 @@ defmodule Ecto.Repo.Schema do
       |> fields_to_sources(dumper)
 
     {rows_or_query, header, placeholder_values, counter} =
-      extract_header_and_fields(repo, rows_or_query, schema, dumper, autogen_id, placeholder_map, adapter, opts)
+      extract_header_and_fields(
+        repo,
+        rows_or_query,
+        schema,
+        dumper,
+        autogen_id,
+        placeholder_map,
+        adapter,
+        opts
+      )
 
     schema_meta = metadata(schema, prefix, source, autogen_id, nil, opts)
 
@@ -55,7 +71,16 @@ defmodule Ecto.Repo.Schema do
     on_conflict = on_conflict(on_conflict, conflict_target, schema_meta, counter, adapter)
 
     {count, rows_or_query} =
-      adapter.insert_all(adapter_meta, schema_meta, header, rows_or_query, on_conflict, return_sources, placeholder_values, opts)
+      adapter.insert_all(
+        adapter_meta,
+        schema_meta,
+        header,
+        rows_or_query,
+        on_conflict,
+        return_sources,
+        placeholder_values,
+        opts
+      )
 
     {count, postprocess(rows_or_query, return_fields_or_types, adapter, schema, schema_meta)}
   end
@@ -77,13 +102,24 @@ defmodule Ecto.Repo.Schema do
     end
   end
 
-  defp extract_header_and_fields(_repo, rows, schema, dumper, autogen_id, placeholder_map, adapter, _opts)
+  defp extract_header_and_fields(
+         _repo,
+         rows,
+         schema,
+         dumper,
+         autogen_id,
+         placeholder_map,
+         adapter,
+         _opts
+       )
        when is_list(rows) do
     mapper = init_mapper(schema, dumper, adapter, placeholder_map)
 
     {rows, {header, has_query?, placeholder_dump, _}} =
       Enum.map_reduce(rows, {%{}, false, %{}, 1}, fn fields, acc ->
-        {fields, {header, has_query?, placeholder_dump, counter}} = Enum.map_reduce(fields, acc, mapper)
+        {fields, {header, has_query?, placeholder_dump, counter}} =
+          Enum.map_reduce(fields, acc, mapper)
+
         {fields, header} = autogenerate_id(autogen_id, fields, header, adapter)
         {fields, {header, has_query?, placeholder_dump, counter}}
       end)
@@ -103,7 +139,7 @@ defmodule Ecto.Repo.Schema do
     placeholder_vals_list =
       placeholder_dump
       |> Enum.map(fn {_, {idx, _, value}} -> {idx, value} end)
-      |> Enum.sort
+      |> Enum.sort()
       |> Enum.map(&elem(&1, 1))
 
     if has_query? do
@@ -114,43 +150,63 @@ defmodule Ecto.Repo.Schema do
     end
   end
 
-  defp extract_header_and_fields(repo, %Ecto.Query{} = query, _schema, _dumper, _autogen_id, _placeholder_map, adapter, opts) do
+  defp extract_header_and_fields(
+         repo,
+         %Ecto.Query{} = query,
+         _schema,
+         _dumper,
+         _autogen_id,
+         _placeholder_map,
+         adapter,
+         opts
+       ) do
     {query, opts} = repo.prepare_query(:insert_all, query, opts)
     query = attach_prefix(query, opts)
 
     {query, params} = Ecto.Adapter.Queryable.plan_query(:insert_all, adapter, query)
 
-    header = case query.select do
-      %Ecto.Query.SelectExpr{expr: {:%{}, _ctx, args}} ->
-        Enum.map(args, &elem(&1, 0))
+    header =
+      case query.select do
+        %Ecto.Query.SelectExpr{expr: {:%{}, _ctx, args}} ->
+          Enum.map(args, &elem(&1, 0))
 
-      _ ->
-        raise ArgumentError, """
-        cannot generate a fields list for insert_all from the given source query
-        because it does not have a select clause that uses a map:
+        _ ->
+          raise ArgumentError, """
+          cannot generate a fields list for insert_all from the given source query
+          because it does not have a select clause that uses a map:
 
-          #{inspect query}
+            #{inspect(query)}
 
-        Please add a select clause that selects into a map, like this:
+          Please add a select clause that selects into a map, like this:
 
-          from x in Source,
-            ...,
-            select: %{
-              field_a: x.bar,
-              field_b: x.foo
-            }
+            from x in Source,
+              ...,
+              select: %{
+                field_a: x.bar,
+                field_b: x.foo
+              }
 
-        The keys must exist in the schema that is being inserted into
-        """
-    end
+          The keys must exist in the schema that is being inserted into
+          """
+      end
 
     counter = fn -> length(params) end
 
     {{query, params}, header, [], counter}
   end
 
-  defp extract_header_and_fields(_repo, rows_or_query, _schema, _dumper, _autogen_id, _placeholder_map, _adapter, _opts) do
-    raise ArgumentError, "expected a list of rows or a query, but got #{inspect rows_or_query} as rows_or_query argument in insert_all"
+  defp extract_header_and_fields(
+         _repo,
+         rows_or_query,
+         _schema,
+         _dumper,
+         _autogen_id,
+         _placeholder_map,
+         _adapter,
+         _opts
+       ) do
+    raise ArgumentError,
+          "expected a list of rows or a query, but got #{inspect(rows_or_query)} as rows_or_query argument in insert_all"
   end
 
   defp init_mapper(nil, _dumper, _adapter, placeholder_map) do
@@ -186,8 +242,7 @@ defmodule Ecto.Repo.Schema do
         {value, placeholder_dump, counter} =
           extract_placeholder(key, type, placeholder_map, placeholder_dump, counter, dumper)
 
-        {{source, value},
-          {Map.put(header, source, true), has_query?, placeholder_dump, counter}}
+        {{source, value}, {Map.put(header, source, true), has_query?, placeholder_dump, counter}}
 
       value ->
         {{source, dumper.(value)},
@@ -360,7 +415,13 @@ defmodule Ecto.Repo.Schema do
           dump_changes!(:insert, Map.take(changes, fields), schema, extra, dumper, adapter)
 
         on_conflict =
-          on_conflict(on_conflict, conflict_target, schema_meta, fn -> length(changes) end, adapter)
+          on_conflict(
+            on_conflict,
+            conflict_target,
+            schema_meta,
+            fn -> length(changes) end,
+            adapter
+          )
 
         args = [adapter_meta, schema_meta, changes, on_conflict, return_sources, opts]
 
@@ -393,9 +454,10 @@ defmodule Ecto.Repo.Schema do
   end
 
   def update(_repo, _name, %{__struct__: _}, opts) when is_list(opts) do
-    raise ArgumentError, "giving a struct to Ecto.Repo.update/2 is not supported. " <>
-                         "Ecto is unable to properly track changes when a struct is given, " <>
-                         "an Ecto.Changeset must be given instead"
+    raise ArgumentError,
+          "giving a struct to Ecto.Repo.update/2 is not supported. " <>
+            "Ecto is unable to properly track changes when a struct is given, " <>
+            "an Ecto.Changeset must be given instead"
   end
 
   defp do_update(repo, name, %Changeset{valid?: true} = changeset, opts) do
@@ -445,13 +507,21 @@ defmodule Ecto.Repo.Schema do
           # If there are no changes or all the changes were autogenerated but not forced, we skip
           {action, autogen} =
             if original != %{} or (autogen != [] and force?),
-               do: {:update, autogen},
-               else: {:noop, []}
+              do: {:update, autogen},
+              else: {:noop, []}
 
           case apply(user_changeset, adapter, action, args) do
             {:ok, values} ->
               changeset
-              |> load_changes(:loaded, return_types, values, embeds, autogen, adapter, schema_meta)
+              |> load_changes(
+                :loaded,
+                return_types,
+                values,
+                embeds,
+                autogen,
+                adapter,
+                schema_meta
+              )
               |> process_children(user_changeset, children, adapter, assoc_opts)
 
             {:error, _} = error ->
@@ -475,10 +545,16 @@ defmodule Ecto.Repo.Schema do
   """
   def insert_or_update(repo, name, changeset, opts) do
     case get_state(changeset) do
-      :built  -> insert(repo, name, changeset, opts)
-      :loaded -> update(repo, name, changeset, opts)
-      state   -> raise ArgumentError, "the changeset has an invalid state " <>
-                                      "for Repo.insert_or_update/2: #{state}"
+      :built ->
+        insert(repo, name, changeset, opts)
+
+      :loaded ->
+        update(repo, name, changeset, opts)
+
+      state ->
+        raise ArgumentError,
+              "the changeset has an invalid state " <>
+                "for Repo.insert_or_update/2: #{state}"
     end
   end
 
@@ -487,18 +563,26 @@ defmodule Ecto.Repo.Schema do
   """
   def insert_or_update!(repo, name, changeset, opts) do
     case get_state(changeset) do
-      :built  -> insert!(repo, name, changeset, opts)
-      :loaded -> update!(repo, name, changeset, opts)
-      state   -> raise ArgumentError, "the changeset has an invalid state " <>
-                                      "for Repo.insert_or_update!/2: #{state}"
+      :built ->
+        insert!(repo, name, changeset, opts)
+
+      :loaded ->
+        update!(repo, name, changeset, opts)
+
+      state ->
+        raise ArgumentError,
+              "the changeset has an invalid state " <>
+                "for Repo.insert_or_update!/2: #{state}"
     end
   end
 
   defp get_state(%Changeset{data: %{__meta__: %{state: state}}}), do: state
+
   defp get_state(%{__struct__: _}) do
-    raise ArgumentError, "giving a struct to Repo.insert_or_update/2 or " <>
-                         "Repo.insert_or_update!/2 is not supported. " <>
-                         "Please use an Ecto.Changeset"
+    raise ArgumentError,
+          "giving a struct to Repo.insert_or_update/2 or " <>
+            "Repo.insert_or_update!/2 is not supported. " <>
+            "Please use an Ecto.Changeset"
   end
 
   @doc """
@@ -541,7 +625,9 @@ defmodule Ecto.Repo.Schema do
 
         case apply(changeset, adapter, :delete, args) do
           {:ok, values} ->
-            changeset = load_changes(changeset, :deleted, [], values, %{}, [], adapter, schema_meta)
+            changeset =
+              load_changes(changeset, :deleted, [], values, %{}, [], adapter, schema_meta)
+
             {:ok, changeset.data}
 
           {:error, _} = error ->
@@ -563,10 +649,13 @@ defmodule Ecto.Repo.Schema do
 
   defp do_load(schema, data, loader) when is_list(data),
     do: do_load(schema, Map.new(data), loader)
+
   defp do_load(schema, {fields, values}, loader) when is_list(fields) and is_list(values),
     do: do_load(schema, Enum.zip(fields, values), loader)
+
   defp do_load(schema, data, loader) when is_atom(schema),
     do: Ecto.Schema.Loader.unsafe_load(schema, data, loader)
+
   defp do_load(types, data, loader) when is_map(types),
     do: Ecto.Schema.Loader.unsafe_load(%{}, types, data, loader)
 
@@ -576,12 +665,17 @@ defmodule Ecto.Repo.Schema do
     case Keyword.get(opts, :returning, false) do
       [_ | _] = fields ->
         fields
+
       [] ->
-        raise ArgumentError, ":returning expects at least one field to be given, got an empty list"
+        raise ArgumentError,
+              ":returning expects at least one field to be given, got an empty list"
+
       true when is_nil(schema) ->
         raise ArgumentError, ":returning option can only be set to true if a schema is given"
+
       true ->
         schema.__schema__(:fields)
+
       false ->
         []
     end
@@ -596,6 +690,7 @@ defmodule Ecto.Repo.Schema do
   defp fields_to_sources(fields, nil) do
     {fields, fields}
   end
+
   defp fields_to_sources(fields, dumper) do
     Enum.reduce(fields, {[], []}, fn field, {types, sources} ->
       {source, type} = Map.fetch!(dumper, field)
@@ -605,20 +700,29 @@ defmodule Ecto.Repo.Schema do
 
   defp struct_from_changeset!(action, %{data: nil}),
     do: raise(ArgumentError, "cannot #{action} a changeset without :data")
+
   defp struct_from_changeset!(_action, %{data: struct}),
     do: struct
 
   defp put_repo_and_action(%{action: :ignore, valid?: valid?} = changeset, action, repo, opts) do
     if valid? do
-      raise ArgumentError, "a valid changeset with action :ignore was given to " <>
-                           "#{inspect repo}.#{action}/2. Changesets can only be ignored " <>
-                           "in a repository action if they are also invalid"
+      raise ArgumentError,
+            "a valid changeset with action :ignore was given to " <>
+              "#{inspect(repo)}.#{action}/2. Changesets can only be ignored " <>
+              "in a repository action if they are also invalid"
     else
       %{changeset | action: action, repo: repo, repo_opts: opts}
     end
   end
-  defp put_repo_and_action(%{action: given}, action, repo, _opts) when given != nil and given != action,
-    do: raise ArgumentError, "a changeset with action #{inspect given} was given to #{inspect repo}.#{action}/2"
+
+  defp put_repo_and_action(%{action: given}, action, repo, _opts)
+       when given != nil and given != action,
+       do:
+         raise(
+           ArgumentError,
+           "a changeset with action #{inspect(given)} was given to #{inspect(repo)}.#{action}/2"
+         )
+
   defp put_repo_and_action(changeset, action, repo, opts),
     do: %{changeset | action: action, repo: repo, repo_opts: opts}
 
@@ -629,8 +733,8 @@ defmodule Ecto.Repo.Schema do
           acc
 
         other ->
-          raise "expected function #{inspect fun} given to Ecto.Changeset.prepare_changes/2 " <>
-                "to return an Ecto.Changeset, got: `#{inspect other}`"
+          raise "expected function #{inspect(fun)} given to Ecto.Changeset.prepare_changes/2 " <>
+                  "to return an Ecto.Changeset, got: `#{inspect(other)}`"
       end
     end)
   end
@@ -644,10 +748,15 @@ defmodule Ecto.Repo.Schema do
       prefix: Keyword.get(opts, :prefix, prefix)
     }
   end
-  defp metadata(%{__struct__: schema, __meta__: %{context: context, source: source, prefix: prefix}},
-                autogen_id, opts) do
+
+  defp metadata(
+         %{__struct__: schema, __meta__: %{context: context, source: source, prefix: prefix}},
+         autogen_id,
+         opts
+       ) do
     metadata(schema, prefix, source, autogen_id, context, opts)
   end
+
   defp metadata(%{__struct__: schema}, _, _) do
     raise ArgumentError, "#{inspect(schema)} needs to be a schema with source"
   end
@@ -655,13 +764,16 @@ defmodule Ecto.Repo.Schema do
   defp conflict_target({:unsafe_fragment, fragment}, _dumper) when is_binary(fragment) do
     {:unsafe_fragment, fragment}
   end
+
   defp conflict_target(conflict_target, dumper) do
     for target <- List.wrap(conflict_target) do
       case dumper do
         %{^target => {alias, _}} ->
           alias
+
         %{} when is_atom(target) ->
           raise ArgumentError, "unknown field `#{inspect(target)}` in conflict_target"
+
         _ ->
           target
       end
@@ -693,14 +805,14 @@ defmodule Ecto.Repo.Schema do
 
       [_ | _] = on_conflict ->
         from = if schema, do: {source, schema}, else: source
-        query = Ecto.Query.from from, update: ^on_conflict
+        query = Ecto.Query.from(from, update: ^on_conflict)
         on_conflict_query(query, {source, schema}, prefix, counter_fun, adapter, conflict_target)
 
       %Ecto.Query{} = query ->
         on_conflict_query(query, {source, schema}, prefix, counter_fun, adapter, conflict_target)
 
       other ->
-        raise ArgumentError, "unknown value for :on_conflict, got: #{inspect other}"
+        raise ArgumentError, "unknown value for :on_conflict, got: #{inspect(other)}"
     end
   end
 
@@ -722,14 +834,14 @@ defmodule Ecto.Repo.Schema do
   end
 
   defp on_conflict_query(query, from, prefix, counter_fun, adapter, conflict_target) do
-    {query, params, _} =
-      Ecto.Query.Planner.plan(%{query | prefix: prefix}, :update_all, adapter)
+    {query, params, _} = Ecto.Query.Planner.plan(%{query | prefix: prefix}, :update_all, adapter)
 
     unless query.from.source == from do
-      raise ArgumentError, "cannot run on_conflict: query because the query " <>
-                           "has a different {source, schema} pair than the " <>
-                           "original struct/changeset/query. Got #{inspect query.from} " <>
-                           "and #{inspect from} respectively"
+      raise ArgumentError,
+            "cannot run on_conflict: query because the query " <>
+              "has a different {source, schema} pair than the " <>
+              "original struct/changeset/query. Got #{inspect(query.from)} " <>
+              "and #{inspect(from)} respectively"
     end
 
     {query, _} = Ecto.Query.Planner.normalize(query, :update_all, adapter, counter_fun.())
@@ -754,7 +866,10 @@ defmodule Ecto.Repo.Schema do
         case Keyword.fetch(opts, :stale_error_field) do
           {:ok, stale_error_field} when is_atom(stale_error_field) ->
             stale_message = Keyword.get(opts, :stale_error_message, "is stale")
-            user_changeset = Changeset.add_error(user_changeset, stale_error_field, stale_message, [stale: true])
+
+            user_changeset =
+              Changeset.add_error(user_changeset, stale_error_field, stale_message, stale: true)
+
             {:error, user_changeset}
 
           _other ->
@@ -763,12 +878,16 @@ defmodule Ecto.Repo.Schema do
     end
   end
 
-  defp constraints_to_errors(%{constraints: user_constraints, errors: errors} = changeset, action, constraints) do
+  defp constraints_to_errors(
+         %{constraints: user_constraints, errors: errors} = changeset,
+         action,
+         constraints
+       ) do
     constraint_errors =
-      Enum.map constraints, fn {type, constraint} ->
+      Enum.map(constraints, fn {type, constraint} ->
         user_constraint =
           Enum.find(user_constraints, fn c ->
-            case {c.type, c.constraint,  c.match} do
+            case {c.type, c.constraint, c.match} do
               {^type, ^constraint, :exact} -> true
               {^type, cc, :suffix} -> String.ends_with?(constraint, cc)
               {^type, cc, :prefix} -> String.starts_with?(constraint, cc)
@@ -779,11 +898,15 @@ defmodule Ecto.Repo.Schema do
         case user_constraint do
           %{field: field, error_message: error_message, error_type: error_type} ->
             {field, {error_message, [constraint: error_type, constraint_name: constraint]}}
+
           nil ->
-            raise Ecto.ConstraintError, action: action, type: type,
-                                        constraint: constraint, changeset: changeset
+            raise Ecto.ConstraintError,
+              action: action,
+              type: type,
+              constraint: constraint,
+              changeset: changeset
         end
-      end
+      end)
 
     %{changeset | errors: constraint_errors ++ errors, valid?: false}
   end
@@ -823,11 +946,14 @@ defmodule Ecto.Repo.Schema do
     case Ecto.Type.adapter_load(adapter, type, value) do
       {:ok, value} ->
         load_each(%{struct | key => value}, kv, types, adapter)
+
       :error ->
-        raise ArgumentError, "cannot load `#{inspect value}` as type #{inspect type} " <>
-                             "for field `#{key}` in schema #{inspect struct.__struct__}"
+        raise ArgumentError,
+              "cannot load `#{inspect(value)}` as type #{inspect(type)} " <>
+                "for field `#{key}` in schema #{inspect(struct.__struct__)}"
     end
   end
+
   defp load_each(struct, [], _types, _adapter) do
     struct
   end
@@ -838,7 +964,7 @@ defmodule Ecto.Repo.Schema do
 
   defp pop_assocs(%{changes: changes, types: types} = changeset, assocs) do
     {changes, parent, child} =
-      Enum.reduce assocs, {changes, [], []}, fn assoc, {changes, parent, child} ->
+      Enum.reduce(assocs, {changes, [], []}, fn assoc, {changes, parent, child} ->
         case changes do
           %{^assoc => value} ->
             changes = Map.delete(changes, assoc)
@@ -846,6 +972,7 @@ defmodule Ecto.Repo.Schema do
             case types do
               %{^assoc => {:assoc, %{relationship: :parent} = refl}} ->
                 {changes, [{refl, value} | parent], child}
+
               %{^assoc => {:assoc, %{relationship: :child} = refl}} ->
                 {changes, parent, [{refl, value} | child]}
             end
@@ -853,7 +980,7 @@ defmodule Ecto.Repo.Schema do
           %{} ->
             {changes, parent, child}
         end
-      end
+      end)
 
     {%{changeset | changes: changes}, parent, child}
   end
@@ -884,21 +1011,25 @@ 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, _}, acc ->
       %{field: field, owner_key: owner_key, related_key: related_key} = refl
       related = Map.get(struct, field)
-      value = related && Map.fetch!(related, related_key)
+      key_mapping = Enum.zip(List.wrap(owner_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}`"
+      Enum.reduce(key_mapping, acc, fn {owner_key, related_key}, acc ->
+        value = related && Map.fetch!(related, related_key)
 
-        _ ->
-          Map.put(acc, owner_key, value)
-      end
-    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(acc, owner_key, value)
+        end
+      end)
+    end)
   end
 
   defp process_children(changeset, user_changeset, assocs, adapter, opts) do
@@ -925,21 +1056,29 @@ defmodule Ecto.Repo.Schema do
 
   defp autogenerate_id({key, source, type}, changes, return_types, return_sources, adapter) do
     cond do
-      Map.has_key?(changes, key) -> # Set by user
+      # Set by user
+      Map.has_key?(changes, key) ->
         {changes, [], return_types, return_sources}
-      value = Ecto.Type.adapter_autogenerate(adapter, type) -> # Autogenerated now
+
+      # Autogenerated now
+      value = Ecto.Type.adapter_autogenerate(adapter, type) ->
         {changes, [{source, value}], [{key, type} | return_types], return_sources}
-      true -> # Autogenerated in storage
-        {changes, [], [{key, type} | return_types], [source | List.delete(return_sources, source)]}
+
+      # Autogenerated in storage
+      true ->
+        {changes, [], [{key, type} | return_types],
+         [source | List.delete(return_sources, source)]}
     end
   end
 
   defp dump_changes!(action, changes, schema, extra, dumper, adapter) do
     autogen = autogenerate_changes(schema, action, changes)
+
     dumped =
       dump_fields!(action, schema, changes, dumper, adapter) ++
-      dump_fields!(action, schema, autogen, dumper, adapter) ++
-      extra
+        dump_fields!(action, schema, autogen, dumper, adapter) ++
+        extra
+
     {dumped, autogen}
   end
 
@@ -962,12 +1101,13 @@ defmodule Ecto.Repo.Schema do
   defp action_to_auto(:update), do: :autoupdate
 
   defp add_pk_filter!(filters, struct) do
-    Enum.reduce Ecto.primary_key!(struct), filters, fn
+    Enum.reduce(Ecto.primary_key!(struct), filters, fn
       {_k, nil}, _acc ->
         raise Ecto.NoPrimaryKeyValueError, struct: struct
+
       {k, v}, acc ->
         Map.put(acc, k, v)
-    end
+    end)
   end
 
   defp wrap_in_transaction(adapter, adapter_meta, opts, changeset, assocs, embeds, prepare, fun) do
@@ -979,8 +1119,8 @@ defmodule Ecto.Repo.Schema do
 
   defp wrap_in_transaction(adapter, adapter_meta, opts, relations_changed?, prepare, fun) do
     if (relations_changed? or prepare != []) and
-       function_exported?(adapter, :transaction, 3) and
-       not adapter.in_transaction?(adapter_meta) do
+         function_exported?(adapter, :transaction, 3) and
+         not adapter.in_transaction?(adapter_meta) do
       adapter.transaction(adapter_meta, opts, fn ->
         case fun.() do
           {:ok, struct} -> struct
@@ -996,10 +1136,11 @@ defmodule Ecto.Repo.Schema do
     case Ecto.Type.adapter_dump(adapter, type, value) do
       {:ok, value} ->
         value
+
       :error ->
         raise Ecto.ChangeError,
               "value `#{inspect(value)}` for `#{inspect(schema)}.#{field}` " <>
-              "in `#{action}` does not match type #{inspect type}"
+                "in `#{action}` does not match type #{inspect(type)}"
     end
   end
 
diff --git a/lib/ecto/schema.ex b/lib/ecto/schema.ex
index 7a5a806ea2..7deef4d917 100644
--- a/lib/ecto/schema.ex
+++ b/lib/ecto/schema.ex
@@ -1976,15 +1976,24 @@ defmodule Ecto.Schema do
   def __belongs_to__(mod, name, queryable, opts) do
     check_options!(opts, @valid_belongs_to_options, "belongs_to/3")
 
-    opts = Keyword.put_new(opts, :foreign_key, :"#{name}_id")
+    opts = Keyword.update(opts, :foreign_key, :"#{name}_id", &List.wrap/1)
     foreign_key_type = opts[:type] || Module.get_attribute(mod, :foreign_key_type)
 
-    if name == Keyword.get(opts, :foreign_key) do
+    if [name] == Keyword.get(opts, :foreign_key) do
       raise ArgumentError, "foreign_key #{inspect name} must be distinct from corresponding association name"
     end
 
     if Keyword.get(opts, :define_field, true) do
-      __field__(mod, opts[:foreign_key], foreign_key_type, opts)
+      foreign_keys = List.wrap(opts[:foreign_key])
+      foreign_key_types = if is_list(foreign_key_type) do
+        foreign_key_type
+      else
+        List.duplicate(foreign_key_type, length(foreign_keys))
+      end
+
+      for {fk, type} <- Enum.zip(foreign_keys, foreign_key_types) do
+        __field__(mod, fk, type, opts)
+      end
     end
 
     struct =
diff --git a/test/ecto/repo/belongs_to_test.exs b/test/ecto/repo/belongs_to_test.exs
index db0d295296..0225f71a46 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 b7758adb39..3995308089 100644
--- a/test/ecto/repo/has_assoc_test.exs
+++ b/test/ecto/repo/has_assoc_test.exs
@@ -563,4 +563,57 @@ defmodule Ecto.Repo.HasAssocTest do
     refute_received {:transaction, _}
     refute_received {:rollback, _}
   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
+
+  # TODO add tests for update
 end
diff --git a/test/ecto/schema_test.exs b/test/ecto/schema_test.exs
index 669e999570..72b2ba59aa 100644
--- a/test/ecto/schema_test.exs
+++ b/test/ecto/schema_test.exs
@@ -564,7 +564,7 @@ defmodule Ecto.SchemaTest do
   test "has_many association" do
     struct =
       %Ecto.Association.Has{field: :posts, owner: AssocSchema, cardinality: :many, on_delete: :nothing,
-                            related: Post, owner_key: :id, related_key: :assoc_schema_id, queryable: Post,
+                            related: Post, owner_key: [:id], related_key: [:assoc_schema_id], queryable: Post,
                             on_replace: :raise}
 
     assert AssocSchema.__schema__(:association, :posts) == struct
@@ -578,7 +578,7 @@ defmodule Ecto.SchemaTest do
   test "has_many association via {source schema}" do
     struct =
       %Ecto.Association.Has{field: :emails, owner: AssocSchema, cardinality: :many, on_delete: :nothing,
-                            related: Email, owner_key: :id, related_key: :assoc_schema_id,
+                            related: Email, owner_key: [:id], related_key: [:assoc_schema_id],
                             queryable: {"users_emails", Email}, on_replace: :delete}
 
     assert AssocSchema.__schema__(:association, :emails) == struct
@@ -592,7 +592,7 @@ defmodule Ecto.SchemaTest do
   test "has_many through association" do
     assert AssocSchema.__schema__(:association, :comment_authors) ==
            %Ecto.Association.HasThrough{field: :comment_authors, owner: AssocSchema, cardinality: :many,
-                                         through: [:comment, :authors], owner_key: :comment_id}
+                                         through: [:comment, :authors], owner_key: [:comment_id]}
 
     refute Map.has_key?(AssocSchema.__changeset__(), :comment_authors)
 
@@ -604,7 +604,7 @@ defmodule Ecto.SchemaTest do
   test "has_one association" do
     struct =
       %Ecto.Association.Has{field: :author, owner: AssocSchema, cardinality: :one, on_delete: :nothing,
-                            related: User, owner_key: :id, related_key: :assoc_schema_id, queryable: User,
+                            related: User, owner_key: [:id], related_key: [:assoc_schema_id], queryable: User,
                             on_replace: :raise}
 
     assert AssocSchema.__schema__(:association, :author) == struct
@@ -618,7 +618,7 @@ defmodule Ecto.SchemaTest do
   test "has_one association via {source, schema}" do
     struct =
       %Ecto.Association.Has{field: :profile, owner: AssocSchema, cardinality: :one, on_delete: :nothing,
-                            related: Profile, owner_key: :id, related_key: :assoc_schema_id,
+                            related: Profile, owner_key: [:id], related_key: [:assoc_schema_id],
                             queryable: {"users_profiles", Profile}, on_replace: :raise}
 
     assert AssocSchema.__schema__(:association, :profile) == struct
@@ -632,7 +632,7 @@ defmodule Ecto.SchemaTest do
   test "has_one through association" do
     assert AssocSchema.__schema__(:association, :comment_main_author) ==
            %Ecto.Association.HasThrough{field: :comment_main_author, owner: AssocSchema, cardinality: :one,
-                                         through: [:comment, :main_author], owner_key: :comment_id}
+                                         through: [:comment, :main_author], owner_key: [:comment_id]}
 
     refute Map.has_key?(AssocSchema.__changeset__(), :comment_main_author)
 
@@ -644,7 +644,7 @@ defmodule Ecto.SchemaTest do
   test "belongs_to association" do
     struct =
       %Ecto.Association.BelongsTo{field: :comment, owner: AssocSchema, cardinality: :one,
-       related: Comment, owner_key: :comment_id, related_key: :id, queryable: Comment,
+       related: Comment, owner_key: [:comment_id], related_key: [:id], queryable: Comment,
        on_replace: :raise, defaults: []}
 
     assert AssocSchema.__schema__(:association, :comment) == struct
@@ -658,7 +658,7 @@ defmodule Ecto.SchemaTest do
   test "belongs_to association via {source, schema}" do
     struct =
       %Ecto.Association.BelongsTo{field: :summary, owner: AssocSchema, cardinality: :one,
-       related: Summary, owner_key: :summary_id, related_key: :id,
+       related: Summary, owner_key: [:summary_id], related_key: [:id],
        queryable: {"post_summary", Summary}, on_replace: :raise, defaults: []}
 
     assert AssocSchema.__schema__(:association, :summary) == struct
@@ -672,7 +672,7 @@ defmodule Ecto.SchemaTest do
   test "belongs_to association via Ecto.ParameterizedType" do
     struct =
       %Ecto.Association.BelongsTo{field: :reference, owner: AssocSchema, cardinality: :one,
-       related: SchemaWithParameterizedPrimaryKey, owner_key: :reference_id, related_key: :id, queryable: SchemaWithParameterizedPrimaryKey,
+       related: SchemaWithParameterizedPrimaryKey, owner_key: [:reference_id], related_key: [:id], queryable: SchemaWithParameterizedPrimaryKey,
        on_replace: :raise, defaults: []}
 
     assert AssocSchema.__schema__(:association, :reference) == struct
@@ -693,32 +693,64 @@ 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 :publication, Publication, references: [:id_1, :id_2],
+        foreign_key: [:publication_id_1, :publication_id_2], type: [:integer, :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
 
   test "has_many options" do
     refl = CustomAssocSchema.__schema__(:association, :posts)
-    assert :pk == refl.owner_key
-    assert :fk == refl.related_key
+    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
+    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
+    # TODO BelongsTo owner_key/related_key should always be lists
     refl = CustomAssocSchema.__schema__(:association, :permalink1)
-    assert :fk == refl.owner_key
-    assert :pk == refl.related_key
+    assert [:fk] == refl.owner_key
+    assert [:pk] == refl.related_key
 
     refl = CustomAssocSchema.__schema__(:association, :permalink2)
-    assert :permalink2_id == refl.owner_key
-    assert :pk == refl.related_key
+    assert [:permalink2_id] == refl.owner_key
+    assert [:pk] == refl.related_key
 
     assert CustomAssocSchema.__schema__(:type, :fk) == :string
     assert CustomAssocSchema.__schema__(:type, :permalink2_id) == :string
+
+    refl = CustomAssocSchema.__schema__(:association, :publication)
+    assert [:publication_id_1, :publication_id_2] == refl.owner_key
+    assert [:id_1, :id_2] == refl.related_key
+
+    assert CustomAssocSchema.__schema__(:type, :publication_id_1) == :integer
+    assert CustomAssocSchema.__schema__(:type, :publication_id_2) == :string
   end
 
   test "has_* validates option" do