Repository: elixir-ecto/ecto
Pull request #4219 (2023-06-27 01:52:29)

Allow `many_to_many` to order preloads by join table
This was inspired by this comment: https://github.com/elixir-ecto/ecto/issues/3509#issuecomment-1606909597.

It seems useful to be able to order by the join table, especially if you are adding metadata to the relationship.

The proposal here was the best one I could think of. It feels a little awkward having two options but the other ideas I thought of didn't seem too great:

1. Mixing the source table option into `preload_order` seems a bit messy.
2. I don't think we want to allow bindings or anything like that in the schema definition. It seems complicated and messy.

josevalim: What about adding `join_preload_order`, to mirror `join_defaults` and `join_where`?
greg-rychlewski: That sounds like a good option to me. Do you think we should raise if both `join_preload_order` and `preload_order` are specified? I think otherwise the order to apply them would be ambiguous.

Maybe this is overkill, but given ^ do you think we should considering giving users the option to order across the tables? Something like: `order by join_table.col_a, assoc_table.col_b, ...`. The API is maybe not so clean though I think you'd have to do a nested keyword list: `[{:join, asc: :col_a}, {:assoc, asc: :col_b}]`
josevalim: How hard would it be to check the columns and, if missing, we assume it is from join?
greg-rychlewski: It would be contained to a relatively small place. Just this function: https://github.com/elixir-ecto/ecto/blob/master/lib/ecto/repo/preloader.ex#L318

```elixir
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)

  [{:asc, related_field} | custom_order_by]
end
```

You have the assoc so you can check if it is many to many. And you have the query so you can get the from source. Then I think you'd just need to use the reflection functions to see if the field exist. Like checking if the type is `nil` or something like that.

The one issue I can see though is you wouldn't be able to distinguish if both tables had the same column name.


josevalim: Yeah, i am not sure what is the best API. :(
josevalim: I would avoid adding them at compile-time. Maybe we could allow preload_order as a MFA and in there you should have access to both?
greg-rychlewski: Sorry if I'm misunderstanding, were you mentioning compile-time due to generating dependencies between modules? 

If so, what I meant was to have "dummy" bindings.  i.e. one fixed binding name for the join and one for the assoc. And then when the option is compiled it will be something like `[{:join, dir1, field}, {:assoc, dir2, field2}, ...]`

I was trying to think of how to use a function too. Though I think we'd have to give them access to modify the entire query rather than generate just the new `order_by` clause. Unless I am not seeing something.
josevalim: Sorry, what I meant is that extracting a macro expression from an option will be tricky and confusing, especially because none of the other options have macro-like behaviour.

> Though I think we'd have to give them access to modify the entire query rather than generate just the new order_by clause. Unless I am not seeing something.

Gah, good point. I guess the best API so far is the one you proposed as `preload_order: [through: [...], assoc: [...]]`. But I would wait a bit more... maybe inspiration will strike later?
greg-rychlewski: Yeah let's wait :). 

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index e4c094ef..d1551b97 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,17 @@
 
   * Adapters now receive `nil` for encoding/decoding
 
+## v3.10.3 (2023-07-07)
+
+### Enhancements
+
+  * [Ecto.Query] Allow dynamic `field/2` in `type/2`
+
+### Bug fixes
+
+  * [Ecto.Changesets] Limit the largest integer to less than 32 digits
+  * [Ecto.Type] Limit the largest integer to less than 32 digits
+
 ## v3.10.2 (2023-06-07)
 
 ### Enhancements
@@ -70,6 +81,17 @@ Queries have also been improved to support LIMIT WITH TIES as well as materializ
   * [Ecto.Query] Fix bug in `json_extract_path` where maps were not allowed to be nested inside of embeds
   * [Ecto.Schema] Allow inline embeds to overwrite conflicting aliases
 
+## v3.9.6 (2023-07-07)
+
+### Enhancements
+
+  * [Ecto.Query] Allow dynamic `field/2` in `type/2`
+
+### Bug fixes
+
+  * [Ecto.Changesets] Limit the largest integer to less than 32 digits
+  * [Ecto.Type] Limit the largest integer to less than 32 digits
+
 ## v3.9.5 (2023-03-22)
 
 ### Bug fixes
diff --git a/lib/ecto/association.ex b/lib/ecto/association.ex
index 0b9ae570..c5d9ad4d 100644
--- a/lib/ecto/association.ex
+++ b/lib/ecto/association.ex
@@ -503,6 +503,10 @@ defmodule Ecto.Association do
   @doc """
   Validates `preload_order` for association named `name`.
   """
+  def validate_preload_order!(_name, {mod, fun, args} = preload_order)
+      when is_atom(mod) and is_atom(fun) and is_list(args),
+      do: preload_order
+
   def validate_preload_order!(name, preload_order) when is_list(preload_order) do
     Enum.map(preload_order, fn
       field when is_atom(field) ->
@@ -528,8 +532,8 @@ defmodule Ecto.Association do
 
   def validate_preload_order!(name, preload_order) do
     raise ArgumentError,
-          "expected `:preload_order` for #{inspect(name)} to be a keyword list or a list of atoms/fields, " <>
-            "got: `#{inspect(preload_order)}`"
+          "expected `:preload_order` for #{inspect(name)} to be a keyword list, a list of atoms/fields " <>
+            "or a {Mod, fun, args} tuple, got: `#{inspect(preload_order)}`"
   end
 
   @doc """
@@ -996,7 +1000,14 @@ defmodule Ecto.Association.Has do
          parent
        ) do
     if value = Map.get(parent, owner_key) do
-      from x in queryable, where: field(x, ^related_key) == ^value
+      query = from x in queryable, where: field(x, ^related_key) == ^value
+
+      parent
+      |> Ecto.get_meta(:prefix)
+      |> case do
+        nil -> query
+        prefix -> Ecto.Query.put_query_prefix(query, prefix)
+      end
     end
   end
 end
@@ -1311,6 +1322,7 @@ defmodule Ecto.Association.ManyToMany do
   @behaviour Ecto.Association
   @on_delete_opts [:nothing, :delete_all]
   @on_replace_opts [:raise, :mark_as_invalid, :delete]
+
   defstruct [
     :field,
     :owner,
diff --git a/lib/ecto/changeset.ex b/lib/ecto/changeset.ex
index 02e2170d..72ab791d 100644
--- a/lib/ecto/changeset.ex
+++ b/lib/ecto/changeset.ex
@@ -1314,7 +1314,9 @@ defmodule Ecto.Changeset do
     end
   end
 
-  defp key_as_int({key, val}) when is_binary(key) do
+  # We check for the byte size to avoid creating unecessary large integers
+  # which would never map to a database key (u64 is 20 digits only).
+  defp key_as_int({key, val}) when is_binary(key) and byte_size(key) < 32 do
     case Integer.parse(key) do
       {key, ""} -> {key, val}
       _ -> {key, val}
@@ -3502,7 +3504,10 @@ defmodule Ecto.Changeset do
 
     * `:message` - the message in case the constraint check fails.
       Defaults to "is invalid"
-    * `:name` - the name of the constraint. Required. It can be a string or a `Regex`
+    * `:name` - the constraint name. By default, the constraint
+      name is inferred from the table + field. If this option is given,
+      the `field` argument only indicates the field the error will be
+      added to. May be required explicitly for complex cases
     * `:match` - how the changeset constraint name is matched against the
       repo constraint, may be `:exact`, `:suffix` or `:prefix`. Defaults to
       `:exact`. `:suffix` matches any repo constraint which `ends_with?` `:name`
@@ -3551,8 +3556,9 @@ defmodule Ecto.Changeset do
       defaults to "has already been taken"
 
     * `:name` - the constraint name. By default, the constraint
-      name is inferred from the table + field(s). May be required
-      explicitly for complex cases
+      name is inferred from the table + field. If this option is given,
+      the `field` argument only indicates the field the error will be
+      added to. May be required explicitly for complex cases
 
     * `:match` - how the changeset constraint name is matched against the
       repo constraint, may be `:exact`, `:suffix` or `:prefix`. Defaults to
@@ -3708,8 +3714,9 @@ defmodule Ecto.Changeset do
     * `:message` - the message in case the constraint check fails,
       defaults to "does not exist"
     * `:name` - the constraint name. By default, the constraint
-      name is inferred from the table + field. May be required
-      explicitly for complex cases
+      name is inferred from the table + field. If this option is given,
+      the `field` argument only indicates the field the error will be
+      added to. May be required explicitly for complex cases
     * `:match` - how the changeset constraint name is matched against the
       repo constraint, may be `:exact`, `:suffix` or `:prefix`. Defaults to
       `:exact`. `:suffix` matches any repo constraint which `ends_with?` `:name`
@@ -3756,8 +3763,9 @@ defmodule Ecto.Changeset do
     * `:message` - the message in case the constraint check fails,
       defaults to "does not exist"
     * `:name` - the constraint name. By default, the constraint
-      name is inferred from the table + association field.
-      May be required explicitly for complex cases
+      name is inferred from the table + field. If this option is given,
+      the `field` argument only indicates the field the error will be
+      added to. May be required explicitly for complex cases
     * `:match` - how the changeset constraint name is matched against the
       repo constraint, may be `:exact`, `:suffix` or `:prefix`. Defaults to
       `:exact`. `:suffix` matches any repo constraint which `ends_with?` `:name`
@@ -3814,8 +3822,9 @@ defmodule Ecto.Changeset do
       defaults to "is still associated with this entry" (for `has_one`)
       and "are still associated with this entry" (for `has_many`)
     * `:name` - the constraint name. By default, the constraint
-      name is inferred from the association table + association
-      field. May be required explicitly for complex cases
+      name is inferred from the table + field. If this option is given,
+      the `field` argument only indicates the field the error will be
+      added to. May be required explicitly for complex cases
     * `:match` - how the changeset constraint name is matched against the
       repo constraint, may be `:exact`, `:suffix` or `:prefix`. Defaults to
       `:exact`. `:suffix` matches any repo constraint which `ends_with?` `:name`
@@ -3857,8 +3866,9 @@ defmodule Ecto.Changeset do
     * `:message` - the message in case the constraint check fails,
       defaults to "violates an exclusion constraint"
     * `:name` - the constraint name. By default, the constraint
-      name is inferred from the table + field. May be required
-      explicitly for complex cases
+      name is inferred from the table + field. If this option is given,
+      the `field` argument only indicates the field the error will be
+      added to. May be required explicitly for complex cases
     * `:match` - how the changeset constraint name is matched against the
       repo constraint, may be `:exact`, `:suffix` or `:prefix`. Defaults to
       `:exact`. `:suffix` matches any repo constraint which `ends_with?` `:name`
diff --git a/lib/ecto/embedded.ex b/lib/ecto/embedded.ex
index 146d409b..7f0099f7 100644
--- a/lib/ecto/embedded.ex
+++ b/lib/ecto/embedded.ex
@@ -259,7 +259,7 @@ defmodule Ecto.Embedded do
       {key, _source, :binary_id} ->
         Map.put_new_lazy(changes, key, fn -> adapter.autogenerate(:embed_id) end)
 
-      {_key, :id} ->
+      {_key, _source, :id} ->
         raise ArgumentError,
               "embedded schema `#{inspect(schema)}` cannot autogenerate `:id` primary keys, " <>
                 "those are typically used for auto-incrementing constraints. " <>
diff --git a/lib/ecto/multi.ex b/lib/ecto/multi.ex
index 31bf8bb9..ba9f6947 100644
--- a/lib/ecto/multi.ex
+++ b/lib/ecto/multi.ex
@@ -148,6 +148,14 @@ defmodule Ecto.Multi do
   5}`.
   """
   @type name :: any
+
+  @typedoc """
+  Result of a failed transaction using a Multi.
+  """
+  @type failure ::
+          {:error, failed_operation :: Ecto.Multi.name(), failed_value :: any(),
+           changes_so_far :: %{Ecto.Multi.name() => any}}
+
   @type t :: %__MODULE__{operations: operations, names: names}
 
   @doc """
diff --git a/lib/ecto/query.ex b/lib/ecto/query.ex
index 140244f3..031af90f 100644
--- a/lib/ecto/query.ex
+++ b/lib/ecto/query.ex
@@ -736,7 +736,7 @@ defmodule Ecto.Query do
 
       windows: [ordered_names: [order_by: e.name]]
 
-  It works exactly as the keyword query version of `order_by/3`.
+  It works exactly as the keyword query version of `order_by/4`.
 
   ### :frame
 
@@ -1364,6 +1364,9 @@ defmodule Ecto.Query do
     * `:materialized` - a boolean indicating whether the CTE should
     be materialized. If blank, the database's default behaviour
     will be used (only supported by Postgrex, for the built-in adapters)
+    * `:operation` - one of `:all`, `:update_all`, or `:delete_all`
+    indicating the operation type of the CTE query. If blank, it defaults to `:all`,
+    making the CTE query a SELECT query. (only supported by Postgres built-in adapter)
 
   ## Recursive CTEs
 
@@ -1431,6 +1434,25 @@ defmodule Ecto.Query do
       |> recursive_ctes(true)
       |> with_cte("category_tree", as: ^category_tree_query)
 
+  For Postgres built-in adapter, it is possible to define data-modifying CTE queries:
+
+      update_categories_query =
+        Category
+        |> where([c], is_nil(c.parent_id))
+        |> update([c], set: [name: "Root category"])
+        |> select([c], c)
+
+      {"update_categories", Category}
+      |> with_cte("update_categories", as: ^update_categories_query, operation: :update_all)
+      |> select([c], c)
+
+  Note: In order to retrieve the updates rows from a CTE query, the parent query
+  must select rows from the CTE table instead of the table referenced by the CTE query.
+  For example, `"update_categories"` will return updates rows for `"category"` table, but
+  selecting from `"category"` table directly will return unaffected rows.
+  For more details see Postgres documentation on data-modifying CTEs and how these work
+  with snapshots.
+
   Keyword syntax is not supported for this feature.
 
   ## Limitation: CTEs on schemas with source fields
@@ -1446,12 +1468,13 @@ defmodule Ecto.Query do
   '''
   defmacro with_cte(query, name, opts) do
     with_query = opts[:as]
+    operation = opts[:operation]
 
     if !with_query do
       Builder.error!("`as` option must be specified")
     end
 
-    Builder.CTE.build(query, name, with_query, opts[:materialized], __CALLER__)
+    Builder.CTE.build(query, name, with_query, opts[:materialized], operation, __CALLER__)
   end
 
   @doc """
@@ -1752,12 +1775,20 @@ defmodule Ecto.Query do
   "nulls first" or "nulls last" is specific to each database implementation.
 
   `order_by` may be invoked or listed in a query many times. New expressions
-  are always appended to the previous ones.
+  can be appended or preprended to the existing ones. The behaviour is controlled
+  through the `:mode` option. By default, new expressions are appended.
 
   `order_by` also accepts a list of atoms where each atom refers to a field in
   source or a keyword list where the direction is given as key and the field
   to order as value.
 
+  ## Options
+
+    * `:mode` - where to place the order expression relative to the
+      ones that already exist. Can be `:append`, to place it after the
+      current orderings, or `:prepend`, to place it before the current
+      orderings. Defaults to `:append`
+
   ## Keywords examples
 
       from(c in City, order_by: c.name, order_by: c.population)
@@ -1806,8 +1837,8 @@ defmodule Ecto.Query do
       City |> order_by(^order_by_param) # Keyword list
 
   """
-  defmacro order_by(query, binding \\ [], expr) do
-    Builder.OrderBy.build(query, binding, expr, __CALLER__)
+  defmacro order_by(query, binding \\ [], expr, opts \\ []) do
+    Builder.OrderBy.build(query, binding, expr, opts, __CALLER__)
   end
 
   @doc """
diff --git a/lib/ecto/query/api.ex b/lib/ecto/query/api.ex
index 8f24a273..58251150 100644
--- a/lib/ecto/query/api.ex
+++ b/lib/ecto/query/api.ex
@@ -421,6 +421,31 @@ defmodule Ecto.Query.API do
   > Because literals are made part of the query, each interpolated
   > literal will generate a separate query, with its own cache.
 
+  ## Splicing
+
+  Sometimes you may need to interpolate a variable number of arguments
+  into the same fragment. For example, when overriding Ecto's default
+  `where` behaviour for Postgres:
+
+      from p in Post, where: fragment("? in (?, ?)", p.id, val1, val2)
+
+  The example above will only work if you know the number of arguments
+  upfront. If it can vary, the above will not work.
+
+  You can address this by telling Ecto to splice a list argument into
+  the fragment:
+
+      from p in Post, where: fragment("? in (?)", p.id, splice(^val_list))
+
+  This will let Ecto know it should expand the values of the list into
+  separate fragment arguments. For example:
+
+      from p in Post, where: fragment("? in (?)", p.id, splice(^[1, 2, 3]))
+
+  would be expanded into
+
+      from p in Post, where: fragment("? in (?,?,?)", p.id, ^1, ^2, ^3)
+
   ## Defining custom functions using macros and fragment
 
   You can add a custom Ecto query function using macros.  For example
@@ -453,6 +478,17 @@ defmodule Ecto.Query.API do
   """
   def fragment(fragments), do: doc! [fragments]
 
+  @doc """
+  Allows a list argument to be spliced into a fragment.
+
+      from p in Post, where: fragment("? in (?)", p.id, splice(^[1, 2, 3]))
+
+  The example above will be transformed at runtime into the following:
+
+      from p in Post, where: fragment("? in (?,?,?)", p.id, ^1, ^2, ^3)
+  """
+  def splice(list), do: doc! [list]
+
   @doc """
   Allows a field to be dynamically accessed.
 
@@ -719,7 +755,7 @@ defmodule Ecto.Query.API do
   referencing it outside of select statements.
 
   This comes in handy when, for instance, you would like to use the calculated
-  value in `Ecto.Query.group_by/3` or `Ecto.Query.order_by/3`:
+  value in `Ecto.Query.group_by/3` or `Ecto.Query.order_by/4`:
 
       from p in Post,
         select: %{
diff --git a/lib/ecto/query/builder.ex b/lib/ecto/query/builder.ex
index 54709086..f74e2672 100644
--- a/lib/ecto/query/builder.ex
+++ b/lib/ecto/query/builder.ex
@@ -74,8 +74,8 @@ defmodule Ecto.Query.Builder do
   with `^index` in the query where index is a number indexing into the
   map.
   """
-  @spec escape(Macro.t, quoted_type | {:in, quoted_type} | {:out, quoted_type}, {list, acc},
-               Keyword.t, Macro.Env.t | {Macro.Env.t, fun}) :: {Macro.t, {list, acc}}
+  @spec escape(Macro.t, quoted_type | {:in, quoted_type} | {:out, quoted_type} | {:splice, quoted_type},
+               {list, acc}, Keyword.t, Macro.Env.t | {Macro.Env.t, fun}) :: {Macro.t, {list, acc}}
   def escape(expr, type, params_acc, vars, env)
 
   # var.x - where var is bound
@@ -694,6 +694,20 @@ defmodule Ecto.Query.Builder do
     end
   end
 
+  defp escape_fragment({:splice, _meta, [splice]}, params_acc, vars, env) do
+    case splice do
+      {:^, _, [value]} = expr ->
+        checked = quote do: Ecto.Query.Builder.splice!(unquote(value))
+        length = quote do: length(unquote(checked))
+        {expr, params_acc} = escape(expr, {:splice, :any}, params_acc, vars, env)
+        escaped =  {:{}, [], [:splice, [], [expr, length]]}
+        {escaped, params_acc}
+
+      _ ->
+        error! "splice/1 in fragment expects an interpolated value, such as splice(^value), got `#{Macro.to_string(splice)}`"
+    end
+  end
+
   defp escape_fragment(expr, params_acc, vars, env) do
     escape(expr, :any, params_acc, vars, env)
   end
@@ -1084,6 +1098,18 @@ defmodule Ecto.Query.Builder do
     end
   end
 
+  @doc """
+  Called by escaper at runtime to verify splice in fragments.
+  """
+  def splice!(value) do
+    if is_list(value) do
+      value
+    else
+      raise ArgumentError,
+            "splice(^value) expects `value` to be a list, got `#{inspect(value)}`"
+    end
+  end
+
   @doc """
   Called by escaper at runtime to verify that value is a valid interval.
   """
diff --git a/lib/ecto/query/builder/cte.ex b/lib/ecto/query/builder/cte.ex
index eb10ebe6..a253fd29 100644
--- a/lib/ecto/query/builder/cte.ex
+++ b/lib/ecto/query/builder/cte.ex
@@ -35,9 +35,9 @@ defmodule Ecto.Query.Builder.CTE do
   If possible, it does all calculations at compile time to avoid
   runtime work.
   """
-  @spec build(Macro.t, Macro.t, Macro.t, nil | boolean(), Macro.Env.t) :: Macro.t
-  def build(query, name, cte, materialized, env) do
-    Builder.apply_query(query, __MODULE__, [escape(name, env), build_cte(name, cte, env), materialized], env)
+  @spec build(Macro.t, Macro.t, Macro.t, nil | boolean(), nil | :all | :update_all | :delete_all , Macro.Env.t) :: Macro.t
+  def build(query, name, cte, materialized, operation, env) do
+    Builder.apply_query(query, __MODULE__, [escape(name, env), build_cte(name, cte, env), materialized, operation], env)
   end
 
   @spec build_cte(Macro.t, Macro.t, Macro.Env.t) :: Macro.t
@@ -73,40 +73,49 @@ defmodule Ecto.Query.Builder.CTE do
   @doc """
   The callback applied by `build/4` to build the query.
   """
-  @spec apply(Ecto.Queryable.t, bitstring, Ecto.Queryable.t, nil | boolean()) :: Ecto.Query.t
+  @spec apply(Ecto.Queryable.t, bitstring, Ecto.Queryable.t, nil | boolean(), nil | :all | :update_all | :delete_all) :: Ecto.Query.t
   # Runtime
-  def apply(%Ecto.Query{with_ctes: with_expr} = query, name, %_{} = with_query, materialized) do
-    %{query | with_ctes: apply_cte(with_expr, name, with_query, materialized)}
+  def apply(query, name, with_query, materialized, nil) do
+    apply(query, name, with_query, materialized, :all)
+  end
+
+  def apply(_query, _name, _with_query, _materialized, operation)
+    when operation not in [:all, :update_all, :delete_all] do
+    Builder.error!("`operation` option must be one of :all, :update_all, or :delete_all")
+  end
+
+  def apply(%Ecto.Query{with_ctes: with_expr} = query, name, %_{} = with_query, materialized, operation) do
+    %{query | with_ctes: apply_cte(with_expr, name, with_query, materialized, operation)}
   end
 
   # Compile
-  def apply(%Ecto.Query{with_ctes: with_expr} = query, name, with_query, materialized) do
+  def apply(%Ecto.Query{with_ctes: with_expr} = query, name, with_query, materialized, operation) do
     update = quote do
-      Ecto.Query.Builder.CTE.apply_cte(unquote(with_expr), unquote(name), unquote(with_query), unquote(materialized))
+      Ecto.Query.Builder.CTE.apply_cte(unquote(with_expr), unquote(name), unquote(with_query), unquote(materialized), unquote(operation))
     end
 
     %{query | with_ctes: update}
   end
 
   # Runtime catch-all
-  def apply(query, name, with_query, materialized) do
-    apply(Ecto.Queryable.to_query(query), name, with_query, materialized)
+  def apply(query, name, with_query, materialized, operation) do
+    apply(Ecto.Queryable.to_query(query), name, with_query, materialized, operation)
   end
 
   @doc false
-  def apply_cte(nil, name, with_query, materialized) when is_boolean(materialized) do
-    %Ecto.Query.WithExpr{queries: [{name, %{materialized: materialized}, with_query}]}
+  def apply_cte(nil, name, with_query, materialized, operation) when is_boolean(materialized) do
+    %Ecto.Query.WithExpr{queries: [{name, %{materialized: materialized, operation: operation}, with_query}]}
   end
 
-  def apply_cte(nil, name, with_query, _materialized)  do
-    %Ecto.Query.WithExpr{queries: [{name, %{}, with_query}]}
+  def apply_cte(nil, name, with_query, _materialized, operation)  do
+    %Ecto.Query.WithExpr{queries: [{name, %{operation: operation}, with_query}]}
   end
 
-  def apply_cte(%Ecto.Query.WithExpr{queries: queries} = with_expr, name, with_query, materialized) when is_boolean(materialized) do
-    %{with_expr | queries:  List.keystore(queries, name, 0, {name, %{materialized: materialized}, with_query})}
+  def apply_cte(%Ecto.Query.WithExpr{queries: queries} = with_expr, name, with_query, materialized, operation) when is_boolean(materialized) do
+    %{with_expr | queries:  List.keystore(queries, name, 0, {name, %{materialized: materialized, operation: operation}, with_query})}
   end
 
-  def apply_cte(%Ecto.Query.WithExpr{queries: queries} = with_expr, name, with_query, _materialized) do
-    %{with_expr | queries:  List.keystore(queries, name, 0, {name, %{}, with_query})}
+  def apply_cte(%Ecto.Query.WithExpr{queries: queries} = with_expr, name, with_query, _materialized, operation) do
+    %{with_expr | queries:  List.keystore(queries, name, 0, {name, %{operation: operation}, with_query})}
   end
 end
diff --git a/lib/ecto/query/builder/order_by.ex b/lib/ecto/query/builder/order_by.ex
index 8a369f58..8c6cf3f1 100644
--- a/lib/ecto/query/builder/order_by.ex
+++ b/lib/ecto/query/builder/order_by.ex
@@ -1,4 +1,4 @@
-import Kernel, except: [apply: 2]
+import Kernel, except: [apply: 3]
 
 defmodule Ecto.Query.Builder.OrderBy do
   @moduledoc false
@@ -14,6 +14,9 @@ defmodule Ecto.Query.Builder.OrderBy do
     :desc_nulls_first
   ]
 
+  @modes [:append, :prepend]
+  @default_mode :append
+
   @doc """
   Returns `true` if term is a valid order_by direction; otherwise returns `false`.
 
@@ -128,6 +131,23 @@ defmodule Ecto.Query.Builder.OrderBy do
 
   defp to_field(field), do: {{:., [], [{:&, [], [0]}, field]}, [], []}
 
+  @doc """
+  Called at runtime to verify the ordering mode
+  """
+  def mode!(order_opts) do
+    case order_opts[:mode] do
+      nil ->
+        @default_mode
+
+      mode when mode in @modes ->
+        mode
+
+      other ->
+        raise ArgumentError,
+              "expected `:mode` to be `:append` or `:prepend`, got: #{inspect(other)}"
+    end
+  end
+
   @doc """
   Shared between order_by and distinct.
   """
@@ -148,10 +168,11 @@ defmodule Ecto.Query.Builder.OrderBy do
   @doc """
   Called at runtime to assemble order_by.
   """
-  def order_by!(query, exprs, file, line) do
+  def order_by!(query, exprs, order_opts, file, line) do
+    mode = Ecto.Query.Builder.OrderBy.mode!(order_opts)
     {expr, params} = order_by_or_distinct!(:order_by, query, exprs, [])
     expr = %Ecto.Query.QueryExpr{expr: expr, params: Enum.reverse(params), line: line, file: file}
-    apply(query, expr)
+    apply(query, expr, mode)
   end
 
   defp dynamic_or_field!(kind, %Ecto.Query.DynamicExpr{} = dynamic, query, {params, count}) do
@@ -176,34 +197,47 @@ defmodule Ecto.Query.Builder.OrderBy do
   If possible, it does all calculations at compile time to avoid
   runtime work.
   """
-  @spec build(Macro.t, [Macro.t], Macro.t, Macro.Env.t) :: Macro.t
-  def build(query, _binding, {:^, _, [var]}, env) do
+  @spec build(Macro.t, [Macro.t], Macro.t, Macro.t, Macro.Env.t) :: Macro.t
+  def build(query, _binding, {:^, _, [var]}, order_opts, env) do
     quote do
-      Ecto.Query.Builder.OrderBy.order_by!(unquote(query), unquote(var), unquote(env.file), unquote(env.line))
+      Ecto.Query.Builder.OrderBy.order_by!(unquote(query), unquote(var), unquote(order_opts), unquote(env.file), unquote(env.line))
     end
   end
 
-  def build(query, binding, expr, env) do
+  def build(query, binding, expr, order_opts, env) do
     {query, binding} = Builder.escape_binding(query, binding, env)
     {expr, {params, _acc}} = escape(:order_by, expr, {[], %{}}, binding, env)
     params = Builder.escape_params(params)
+    mode = quote do: Ecto.Query.Builder.OrderBy.mode!(unquote(order_opts))
 
     order_by = quote do: %Ecto.Query.QueryExpr{
                            expr: unquote(expr),
                            params: unquote(params),
                            file: unquote(env.file),
                            line: unquote(env.line)}
-    Builder.apply_query(query, __MODULE__, [order_by], env)
+    Builder.apply_query(query, __MODULE__, [order_by, mode], env)
   end
 
   @doc """
   The callback applied by `build/4` to build the query.
   """
-  @spec apply(Ecto.Queryable.t, term) :: Ecto.Query.t
-  def apply(%Ecto.Query{order_bys: order_bys} = query, expr) do
-    %{query | order_bys: order_bys ++ [expr]}
+  @spec apply(Ecto.Queryable.t, term, term) :: Ecto.Query.t
+  def apply(%Ecto.Query{order_bys: orders} = query, expr, mode) do
+    %{query | order_bys: update_order_bys(orders, expr, mode)}
   end
-  def apply(query, expr) do
-    apply(Ecto.Queryable.to_query(query), expr)
+  def apply(query, expr, mode) do
+    apply(Ecto.Queryable.to_query(query), expr, mode)
+  end
+
+  @doc """
+  Updates the `order_bys` value for a query.
+  """
+  def update_order_bys(orders, expr, :append), do: orders ++ [expr]
+  def update_order_bys(orders, expr, :prepend), do: [expr | orders]
+
+  def update_order_bys(orders, expr, mode) do
+    quote do
+      Ecto.Query.Builder.OrderBy.update_order_bys(unquote(orders), unquote(expr), unquote(mode))
+    end
   end
 end
diff --git a/lib/ecto/query/planner.ex b/lib/ecto/query/planner.ex
index 261e706e..6f816d55 100644
--- a/lib/ecto/query/planner.ex
+++ b/lib/ecto/query/planner.ex
@@ -757,11 +757,11 @@ defmodule Ecto.Query.Planner do
     Enum.reduce queries, cache_and_params, fn
       {name, opts, %Ecto.Query{} = query}, {cache, params} ->
         {_, params, inner_cache} = traverse_cache(query, :all, {[], params}, adapter)
-        {merge_cache({key, name, opts[:materialized], inner_cache}, cache, inner_cache != :nocache), params}
+        {merge_cache({key, name, opts[:materialized], opts[:operation], inner_cache}, cache, inner_cache != :nocache), params}
 
       {name, opts, %Ecto.Query.QueryExpr{} = query_expr}, {cache, params} ->
         {params, cacheable?} = cast_and_merge_params(:with_cte, query, query_expr, params, adapter)
-        {merge_cache({key, name, opts[:materialized], expr_to_cache(query_expr)}, cache, cacheable?), params}
+        {merge_cache({key, name, opts[:materialized], opts[:operation], expr_to_cache(query_expr)}, cache, cacheable?), params}
     end
   end
 
@@ -785,13 +785,14 @@ defmodule Ecto.Query.Planner do
 
       {v, type}, {acc, cacheable?} ->
         case cast_param(kind, query, expr, v, type, adapter) do
-          {cast_v, {:in, dump_v}} -> {split_in_params(cast_v, dump_v, acc), false}
+          {cast_v, {:in, dump_v}} -> {split_variadic_params(cast_v, dump_v, acc), false}
+          {cast_v, {:splice, dump_v}} -> {split_variadic_params(cast_v, dump_v, acc), cacheable?}
           cast_v_and_dump_v -> {[cast_v_and_dump_v | acc], cacheable?}
         end
     end
   end
 
-  defp split_in_params(cast_v, dump_v, acc) do
+  defp split_variadic_params(cast_v, dump_v, acc) do
     Enum.zip(cast_v, dump_v) |> Enum.reverse(acc)
   end
 
@@ -1051,7 +1052,7 @@ defmodule Ecto.Query.Planner do
           # We don't want to use normalize_subquery_select because we are
           # going to prepare the whole query ourselves next.
           {_, _, inner_query} = rewrite_subquery_select_expr(inner_query, true)
-          {inner_query, counter} = traverse_exprs(inner_query, :all, counter, fun)
+          {inner_query, counter} = traverse_exprs(inner_query, opts.operation, counter, fun)
 
           # Now compute the fields as keyword lists so we emit AS in Ecto query.
           %{select: %{expr: expr, take: take, aliases: aliases}} = inner_query
@@ -1248,6 +1249,11 @@ defmodule Ecto.Query.Planner do
     {{quantifier, meta, [subquery]}, acc}
   end
 
+  defp prewalk({:splice, splice_meta, [{:^, meta, [_]}, length]}, _kind, _query, _expr, acc, _adapter) do
+    param = {:^, meta, [acc, length]}
+    {{:splice, splice_meta, [param]}, acc + length}
+  end
+
   defp prewalk({{:., dot_meta, [left, field]}, meta, []},
                kind, query, expr, acc, _adapter) do
     {ix, ix_expr, ix_query} = get_ix!(left, kind, query)
diff --git a/lib/ecto/repo.ex b/lib/ecto/repo.ex
index e5d4dd35..316b8d8b 100644
--- a/lib/ecto/repo.ex
+++ b/lib/ecto/repo.ex
@@ -99,6 +99,11 @@ defmodule Ecto.Repo do
     * `:telemetry_options` - Extra options to attach to telemetry event name.
       See the next section for more information
 
+  ## Adapter-Specific Errors
+
+  Many of the functions in this module may raise adapter-specific errors, such as `PostgrexError`.
+  This can happen, for example, when the underlying database cannot execute the specified query.
+
   ## Telemetry events
 
   There are two types of telemetry events. The ones emitted by Ecto and the
@@ -2100,7 +2105,7 @@ defmodule Ecto.Repo do
   @callback transaction(fun_or_multi :: fun | Ecto.Multi.t(), opts :: Keyword.t()) ::
               {:ok, any}
               | {:error, any}
-              | {:error, Ecto.Multi.name(), any, %{Ecto.Multi.name() => any}}
+              | Ecto.Multi.failure()
 
   @doc """
   Returns true if the current process is inside a transaction.
diff --git a/lib/ecto/repo/preloader.ex b/lib/ecto/repo/preloader.ex
index e982d3df..db3b41d8 100644
--- a/lib/ecto/repo/preloader.ex
+++ b/lib/ecto/repo/preloader.ex
@@ -6,6 +6,8 @@ defmodule Ecto.Repo.Preloader do
   require Ecto.Query
   require Logger
 
+  alias Ecto.Query.DynamicExpr
+
   @doc """
   Transforms a result set based on query preloads, loading
   the associations onto their parent schema.
@@ -242,13 +244,13 @@ 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)
+    related_field_ast = related_key_to_field(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, &{:{}, [], [related_field_ast, &1]}
 
     # If we are returning many results, we must sort by the key too
     query =
@@ -261,10 +263,13 @@ defmodule Ecto.Repo.Preloader do
                   "select the parent's foreign key"
 
         {:many, _} ->
+          query = add_preload_order(assoc.preload_order, query)
+
           update_in query.order_bys, fn order_bys ->
-            [%Ecto.Query.QueryExpr{expr: preload_order(assoc, query, field), params: [],
+            [%Ecto.Query.QueryExpr{expr: [asc: related_field_ast], params: [],
                                    file: __ENV__.file, line: __ENV__.line}|order_bys]
           end
+
         {:one, _} ->
           query
       end
@@ -315,17 +320,6 @@ defmodule Ecto.Repo.Preloader do
     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)
-
-    [{:asc, related_field} | custom_order_by]
-  end
-
   defp related_key_to_field(query, {pos, key, field_type}) do
     field_ast = related_key_to_field(query, {pos, key})
 
@@ -339,6 +333,46 @@ defmodule Ecto.Repo.Preloader do
   defp related_key_pos(_query, pos) when pos >= 0, do: pos
   defp related_key_pos(query, pos), do: Ecto.Query.Builder.count_binds(query) + pos
 
+  defp add_preload_order([], query), do: query
+
+  defp add_preload_order(order, query) when is_list(order) do
+    Ecto.Query.order_by(query, [q], ^order, mode: :prepend)
+  end
+
+  defp add_preload_order({m, f, a}, query) do
+    order =
+      case apply(m, f, a) do
+        order when is_list(order) ->
+          order
+
+        other ->
+          raise ArgumentError,
+                "`:preload_order` must resolve to a keyword list or a list of atoms/fields, " <>
+                  "got: `#{inspect(other)}`"
+      end
+
+    Enum.each(order, fn
+      {direction, field} when is_atom(field) or is_struct(field, DynamicExpr) ->
+        unless Ecto.Query.Builder.OrderBy.valid_direction?(direction) do
+          raise ArgumentError,
+                "`:preload_order` must specify valid directions, " <>
+                  "got: `#{inspect(order)}`, `#{inspect(direction)}` is not a valid direction"
+        end
+
+        :ok
+
+      field when is_atom(field) or is_struct(field, DynamicExpr) ->
+        :ok
+
+      other ->
+        raise ArgumentError,
+              "`:preload_order` must resolve to a keyword list or a list of atoms/fields, " <>
+                "got: `#{inspect(order)}`, `#{inspect(other)}` is not valid"
+    end)
+
+    add_preload_order(order, query)
+  end
+
   defp unzip_ids([{k, v}|t], acc1, acc2), do: unzip_ids(t, [k|acc1], [v|acc2])
   defp unzip_ids([], acc1, acc2), do: {acc1, acc2}
 
diff --git a/lib/ecto/schema.ex b/lib/ecto/schema.ex
index a623bcfe..6add8e14 100644
--- a/lib/ecto/schema.ex
+++ b/lib/ecto/schema.ex
@@ -831,7 +831,7 @@ defmodule Ecto.Schema do
       For example, if you set `Post.has_many :comments, preload_order: [asc: :content]`,
       whenever the `:comments` associations is preloaded,
       the comments will be order by the `:content` field.
-      See `Ecto.Query.order_by/3` for more examples.
+      See `Ecto.Query.order_by/4` for more examples.
 
   ## Examples
 
@@ -1369,11 +1369,12 @@ defmodule Ecto.Schema do
     * `:join_where` - A filter for the join table. See "Filtering associations"
       in `has_many/3`
 
-    * `:preload_order` - Sets the default `order_by` of the association.
-      It is used when the association is preloaded.
-      For example, if you set `Post.many_to_many :tags, Tag, join_through: "posts_tags", preload_order: [asc: :foo]`,
-      whenever the `:tags` associations is preloaded, the tags will be order by the `:foo` field.
-      See `Ecto.Query.order_by/3` for more examples.
+    * `:preload_order` - Sets the default `order_by` when preloading the association.
+      It may be a keyword list/list of fields or an MFA tuple, such as `{Mod, fun, []}`.
+      Both cases must resolve to a valid `order_by` expression. See `Ecto.Query.order_by/4`
+      to learn more about about ordering expressions.
+      See the [preload order](#many_to_many/3-preload-order) section below to learn how
+      this option can be utilized
 
   ## Using Ecto.assoc/2
 
@@ -1445,6 +1446,46 @@ defmodule Ecto.Schema do
   Because `:join_through` contains a schema, in such cases, autogenerated
   values and primary keys will be automatically handled by Ecto.
 
+  ## Preload Order
+
+  The `:preload_order` option may be used to return the preloaded structs
+  in a deterministic order. It accepts either a compile-time keyword list/list
+  or an MFA tuple, such as `{Mod, fun, []}`. The MFA tuple will be used to
+  generate the `order_by` expression at runtime.
+
+  When specifying a compile-time keyword list/list, the ordering applies to the
+  association's table and not the join table. Ordering by the join table can be
+  achieved by specifying an MFA tuple that utilizes `Ecto.Query.dynamic/2`.
+
+  For example, say we have an association `Assoc` being joined through the table
+  `join_through`. The default preload query generated by Ecto is roughly:
+
+      from a in Assoc, join: jt in "join_through", on: ...
+
+  If `:preload_order` is given as `[asc: :field]` then the preload query will be
+  changed to the following:
+
+      from a in Assoc, join: jt in "join_through", on: ..., order_by: [asc: a.field]
+
+  Similarly, any compile-time keyword list/list will have its fields interpreted
+  as belonging to the association's table. To order by a field from the join table,
+  an MFA tuple can be specified that utilizes `Ecto.Query.dynamic/2`.
+
+  For example, if `:preload_order` is given as `{Mod, fun, []}`, corresponding to
+  the following function:
+
+      defmodule Mod do
+        def fun() do
+          [desc: dynamic([assoc, join], join.field)]
+        end
+      end
+
+  then the preload query will be changed to the following:
+
+      from a in Assoc, join: jt in "join_through", on: ..., order_by: [desc: jt.field]
+
+  Note the ordering of the bindings. The join table always comes last.
+
   ## Examples
 
       defmodule Post do
diff --git a/lib/ecto/type.ex b/lib/ecto/type.ex
index 99d4b9e2..991b9ebc 100644
--- a/lib/ecto/type.ex
+++ b/lib/ecto/type.ex
@@ -211,6 +211,7 @@ defmodule Ecto.Type do
     utc_datetime_usec naive_datetime_usec time_usec
   )a
   @composite ~w(array map maybe in param)a
+  @variadic ~w(in splice)a
 
   @doc """
   Returns the underlying schema type for the custom type.
@@ -507,9 +508,9 @@ defmodule Ecto.Type do
     end
   end
 
-  def dump({:in, type}, value, dumper) do
+  def dump({qual, type}, value, dumper) when qual in @variadic do
     case dump({:array, type}, value, dumper) do
-      {:ok, value} -> {:ok, {:in, value}}
+      {:ok, value} -> {:ok, {qual, value}}
       :error -> :error
     end
   end
@@ -795,7 +796,7 @@ defmodule Ecto.Type do
   defp cast_fun(:utc_datetime_usec), do: &maybe_pad_usec(cast_utc_datetime(&1))
   defp cast_fun({:param, :any_datetime}), do: &cast_any_datetime(&1)
   defp cast_fun({:parameterized, mod, params}), do: &mod.cast(&1, params)
-  defp cast_fun({:in, type}), do: cast_fun({:array, type})
+  defp cast_fun({qual, type}) when qual in @variadic, do: cast_fun({:array, type})
 
   defp cast_fun({:array, {:parameterized, _, _} = type}) do
     fun = cast_fun(type)
@@ -824,7 +825,9 @@ defmodule Ecto.Type do
     end
   end
 
-  defp cast_integer(term) when is_binary(term) do
+  # We check for the byte size to avoid creating unecessary large integers
+  # which would never map to a database key (u64 is 20 digits only).
+  defp cast_integer(term) when is_binary(term) and byte_size(term) < 32 do
     case Integer.parse(term) do
       {integer, ""} -> {:ok, integer}
       _ -> :error
@@ -1410,16 +1413,16 @@ defmodule Ecto.Type do
     {:ok, acc}
   end
 
-  defp to_i(nil), do: nil
-  defp to_i(int) when is_integer(int), do: int
-
-  defp to_i(bin) when is_binary(bin) do
+  defp to_i(bin) when is_binary(bin) and byte_size(bin) < 32 do
     case Integer.parse(bin) do
       {int, ""} -> int
       _ -> nil
     end
   end
 
+  defp to_i(int) when is_integer(int), do: int
+  defp to_i(_), do: nil
+
   defp maybe_truncate_usec({:ok, struct}), do: {:ok, truncate_usec(struct)}
   defp maybe_truncate_usec(:error), do: :error
 


Test patch:
diff --git a/integration_test/cases/preload.exs b/integration_test/cases/preload.exs
index f5046799..8cd872ba 100644
--- a/integration_test/cases/preload.exs
+++ b/integration_test/cases/preload.exs
@@ -575,6 +575,16 @@ defmodule Ecto.Integration.PreloadTest do
     assert [%{name: "foo"}, %{name: "bar"}] = post.ordered_users
   end
 
+  test "custom preload_order with mfa" do
+    post1 = TestRepo.insert!(%Post{users: [%User{name: "bar"}, %User{name: "foo"}], title: "1"})
+    post2 = TestRepo.insert!(%Post{users: [%User{name: "baz"}, %User{name: "foz"}], title: "2"})
+
+    [post1, post2] = TestRepo.preload([post1, post2], [:ordered_users_by_join_table], log: :error)
+
+    assert [%{name: "foo"}, %{name: "bar"}] = post1.ordered_users_by_join_table
+    assert [%{name: "foz"}, %{name: "baz"}] = post2.ordered_users_by_join_table
+  end
+
   ## Others
 
   @tag :invalid_prefix
diff --git a/integration_test/support/schemas.exs b/integration_test/support/schemas.exs
index ea646519..94ddbfab 100644
--- a/integration_test/support/schemas.exs
+++ b/integration_test/support/schemas.exs
@@ -28,6 +28,7 @@ defmodule Ecto.Integration.Post do
   """
   use Ecto.Integration.Schema
   import Ecto.Changeset
+  import Ecto.Query, only: [dynamic: 2]
 
   schema "posts" do
     field :counter, :id # Same as integer
@@ -57,6 +58,8 @@ defmodule Ecto.Integration.Post do
     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]
+    many_to_many :ordered_users_by_join_table, Ecto.Integration.User,
+      join_through: "posts_users", preload_order: {__MODULE__, :preload_order, []}
     many_to_many :unique_users, Ecto.Integration.User,
       join_through: "posts_users", unique: true
     many_to_many :constraint_users, Ecto.Integration.User,
@@ -67,6 +70,10 @@ defmodule Ecto.Integration.Post do
     timestamps()
   end
 
+  def preload_order() do
+    [desc: dynamic([assoc, join], join.user_id)]
+  end
+
   def changeset(schema, params) do
     cast(schema, params, ~w(counter title blob temp public cost visits
                            intensity bid uuid meta posted)a)
diff --git a/test/ecto/query/builder/order_by_test.exs b/test/ecto/query/builder/order_by_test.exs
index 6c822c56..da76e079 100644
--- a/test/ecto/query/builder/order_by_test.exs
+++ b/test/ecto/query/builder/order_by_test.exs
@@ -88,6 +88,36 @@ defmodule Ecto.Query.Builder.OrderByTest do
         escape(:order_by, quote do [desc: selected_as("ident")] end, {[], %{}}, [], __ENV__)
       end
     end
+
+    test "accepts :mode option" do
+      query = order_by("q", [q], [:title])
+      opts = [mode: :prepend]
+      %{order_bys: prepend_order_bys} = order_by(query, [q], [:prepend], opts)
+      %{order_bys: append_order_bys} = order_by(query, [q], [:append], mode: :append)
+      %{order_bys: default_order_bys} = order_by(query, [q], [:append])
+
+      assert [
+               %{expr: [asc: {{:., _, [_, :prepend]}, _, _}]},
+               %{expr: [asc: {{:., _, [_, :title]}, _, _}]}
+             ] = prepend_order_bys
+
+      assert [
+               %{expr: [asc: {{:., _, [_, :title]}, _, _}]},
+               %{expr: [asc: {{:., _, [_, :append]}, _, _}]}
+             ] = append_order_bys
+
+      assert [
+               %{expr: [asc: {{:., _, [_, :title]}, _, _}]},
+               %{expr: [asc: {{:., _, [_, :append]}, _, _}]}
+             ] = default_order_bys
+
+
+      error_msg = "expected `:mode` to be `:append` or `:prepend`, got: :invalid"
+
+      assert_raise ArgumentError, error_msg, fn ->
+        order_by(query, [q], [:invalid], mode: :invalid)
+      end
+    end
   end
 
   describe "at runtime" do
@@ -100,6 +130,36 @@ defmodule Ecto.Query.Builder.OrderByTest do
       assert order_by("q", [q], [{^dir, ^key}]).order_bys == order_by("q", [q], [desc: q.title]).order_bys
     end
 
+    test "accepts :mode option" do
+      query = order_by("q", [q], [:title])
+      opts = [mode: :prepend]
+      %{order_bys: prepend_order_bys} = order_by(query, [q], ^[:prepend], opts)
+      %{order_bys: append_order_bys} = order_by(query, [q], ^[:append], mode: :append)
+      %{order_bys: default_order_bys} = order_by(query, [q], ^[:append])
+
+      assert [
+               %{expr: [asc: {{:., _, [_, :prepend]}, _, _}]},
+               %{expr: [asc: {{:., _, [_, :title]}, _, _}]}
+             ] = prepend_order_bys
+
+      assert [
+               %{expr: [asc: {{:., _, [_, :title]}, _, _}]},
+               %{expr: [asc: {{:., _, [_, :append]}, _, _}]}
+             ] = append_order_bys
+
+      assert [
+               %{expr: [asc: {{:., _, [_, :title]}, _, _}]},
+               %{expr: [asc: {{:., _, [_, :append]}, _, _}]}
+             ] = default_order_bys
+
+
+      error_msg = "expected `:mode` to be `:append` or `:prepend`, got: :invalid"
+
+      assert_raise ArgumentError, error_msg, fn ->
+        order_by(query, [q], ^[:invalid], mode: :invalid)
+      end
+    end
+
     test "supports dynamic expressions" do
       order_by = [
         asc: dynamic([p], p.foo == ^1 and p.bar == ^"bar"),
diff --git a/test/ecto/query/planner_test.exs b/test/ecto/query/planner_test.exs
index 74cc6da0..63497413 100644
--- a/test/ecto/query/planner_test.exs
+++ b/test/ecto/query/planner_test.exs
@@ -738,8 +738,7 @@ defmodule Ecto.Query.PlannerTest do
       assert [
         :all,
         {:from, {"comments", Comment, _, nil}, []},
-        {:non_recursive_cte, "cte",
-        nil,
+        {:non_recursive_cte, "cte", nil, :all,
          [:all, {:prefix, "another"}, {:from, {"comments", Comment, _, nil}, []}, {:select, {:&, _, [0]}}]}
       ] = cache
 
@@ -760,7 +759,7 @@ defmodule Ecto.Query.PlannerTest do
       %{queries: [{"cte", %{}, query_expr}]} = with_expr
       expr = {:fragment, [], [raw: "SELECT * FROM comments WHERE id = ", expr: {:^, [], [0]}, raw: ""]}
       assert expr == query_expr.expr
-      assert [:all, {:from, {"comments", Comment, _, nil}, []}, {:recursive_cte, "cte", nil, ^expr}] = cache
+      assert [:all, {:from, {"comments", Comment, _, nil}, []}, {:recursive_cte, "cte", nil, :all, ^expr}] = cache
     end
 
     test "on update_all" do
@@ -784,7 +783,7 @@ defmodule Ecto.Query.PlannerTest do
       assert {{"comments", Comment, "another"}} = cte.sources
       assert %{expr: {:^, [], [0]}, params: [{500, :integer}]} = cte.limit
 
-      assert [:update_all, _, _, _, _, {:non_recursive_cte, "recent_comments", nil, cte_cache}] = cache
+      assert [:update_all, _, _, _, _, {:non_recursive_cte, "recent_comments", nil, :all, cte_cache}] = cache
       assert [
                :all,
                {:prefix, "another"},
@@ -816,7 +815,7 @@ defmodule Ecto.Query.PlannerTest do
       assert {{"comments", Comment, "another"}} = cte.sources
       assert %{expr: {:^, [], [0]}, params: [{500, :integer}]} = cte.limit
 
-      assert [:delete_all, _, _, _, {:non_recursive_cte, "recent_comments", nil, cte_cache}] = cache
+      assert [:delete_all, _, _, _, {:non_recursive_cte, "recent_comments", nil, :all, cte_cache}] = cache
       assert [
                :all,
                {:prefix, "another"},
@@ -1355,6 +1354,24 @@ defmodule Ecto.Query.PlannerTest do
     assert query.group_bys == []
   end
 
+  test "normalize: fragment with splicing" do
+    two = 2
+    three = 3
+
+    {query, cast_params, dump_params, _} =
+      from(c in Comment)
+      |> where([c], c.id in fragment("(?, ?, ?)", ^1, splice(^[two, three, 4]), ^5))
+      |> normalize_with_params()
+
+    assert cast_params == [1, 2, 3, 4, 5]
+    assert dump_params == [1, 2, 3, 4, 5]
+
+    {:in, _, [_, {:fragment, _, parts}]} = hd(query.wheres).expr
+    assert [_, _, _, {:expr, {:splice, _, [{:^, _, [start_ix, length]}]}}, _, _, _] = parts
+    assert start_ix == 1
+    assert length == 3
+  end
+
   describe "normalize: CTEs" do
     test "single-level" do
       %{with_ctes: with_expr} =
diff --git a/test/ecto/query_test.exs b/test/ecto/query_test.exs
index df90e37d..d1bbdbef 100644
--- a/test/ecto/query_test.exs
+++ b/test/ecto/query_test.exs
@@ -780,10 +780,10 @@ defmodule Ecto.QueryTest do
 
       excluded_left_lateral_query = exclude(left_lateral_query, :left_lateral_join)
       assert excluded_left_lateral_query.joins == base.joins
-      
+
       excluded_array_query = exclude(array_query, :array_join)
       assert excluded_array_query.joins == base.joins
-      
+
       excluded_left_array_query = exclude(left_array_query, :left_array_join)
       assert excluded_left_array_query.joins == base.joins
     end
@@ -868,13 +868,13 @@ defmodule Ecto.QueryTest do
       assert map_size(excluded_left_lateral_join_query.aliases) == original_aliases_number - 1
       refute Map.has_key?(excluded_left_lateral_join_query.aliases, :blogs_ll)
       assert Map.has_key?(excluded_left_lateral_join_query.aliases, :base)
-      
+
       excluded_array_join_query = exclude(query, :array_join)
       assert length(excluded_array_join_query.joins) == original_joins_number - 1
       assert map_size(excluded_array_join_query.aliases) == original_aliases_number - 1
       refute Map.has_key?(excluded_array_join_query.aliases, :blogs_a)
       assert Map.has_key?(excluded_array_join_query.aliases, :base)
-      
+
       excluded_left_array_join_query = exclude(query, :left_array_join)
       assert length(excluded_left_array_join_query.joins) == original_joins_number - 1
       assert map_size(excluded_left_array_join_query.aliases) == original_aliases_number - 1
@@ -991,6 +991,30 @@ defmodule Ecto.QueryTest do
       end
     end
 
+    test "supports list splicing" do
+      two = 2
+      three = 3
+
+      query =
+        from p in "posts", where: p.id in fragment("(?, ?, ?)", ^1, splice(^[two, three, 4]), ^5)
+
+      assert {:in, _, [_, {:fragment, _, parts}]} = hd(query.wheres).expr
+
+      assert [
+               raw: "(",
+               expr: {:^, _, [0]},
+               raw: ", ",
+               expr: {:splice, _, [{:^, _, [1]}, 3]},
+               raw: ", ",
+               expr: {:^, _, [2]},
+               raw: ")"
+             ] = parts
+
+      assert_raise ArgumentError, "splice(^value) expects `value` to be a list, got `234`", fn ->
+        from p in "posts", where: p.id in fragment("(?)", splice(^234))
+      end
+    end
+
     test "keeps UTF-8 encoding" do
       assert inspect(from p in "posts", where: fragment("héllò")) ==
              ~s[#Ecto.Query<from p0 in \"posts\", where: fragment("héllò")>]
diff --git a/test/ecto/repo/embedded_test.exs b/test/ecto/repo/embedded_test.exs
index f942cb69..a9dc02d4 100644
--- a/test/ecto/repo/embedded_test.exs
+++ b/test/ecto/repo/embedded_test.exs
@@ -163,6 +163,31 @@ defmodule Ecto.Repo.EmbeddedTest do
     assert errors == %{embeds: [%{}, %{id: ["has already been taken"]}]}
   end
 
+  test "raise when trying to autogenerate embed primary keys with type `:id`" do
+    defmodule EmbedWithPrimaryKeyTypeId do
+      use Ecto.Schema
+
+      @primary_key {:id, :id, autogenerate: true}
+      embedded_schema do
+      end
+    end
+
+    defmodule SchemaWithPrimaryKeyTypeIdEmbed do
+      use Ecto.Schema
+
+      schema "" do
+        embeds_one :embed, EmbedWithPrimaryKeyTypeId
+      end
+    end
+
+    embed = struct(EmbedWithPrimaryKeyTypeId)
+    schema = struct(SchemaWithPrimaryKeyTypeIdEmbed, embed: embed)
+
+    assert_raise ArgumentError, ~r/cannot autogenerate `:id` primary keys/, fn ->
+      TestRepo.insert!(schema)
+    end
+  end
+
   ## update
 
   test "skips embeds on update when not changing" do
diff --git a/test/ecto/repo/has_assoc_test.exs b/test/ecto/repo/has_assoc_test.exs
index b2af1efe..48831c8c 100644
--- a/test/ecto/repo/has_assoc_test.exs
+++ b/test/ecto/repo/has_assoc_test.exs
@@ -41,11 +41,51 @@ defmodule Ecto.Repo.HasAssocTest do
       has_one :assoc, MyAssoc, on_replace: :delete
       has_one :nilify_assoc, MyAssoc, on_replace: :nilify
       has_one :delete_assoc, MyAssoc
+      has_one :on_delete_delete_assoc, MyAssoc, on_delete: :delete_all
+      has_one :on_delete_nilify_assoc, MyAssoc, on_delete: :nilify_all
       has_many :assocs, MyAssoc, on_replace: :delete
       has_many :delete_assocs, MyAssoc
+      has_many :on_delete_delete_assocs, MyAssoc, on_delete: :delete_all
+      has_many :on_delete_nilify_assocs, MyAssoc, on_delete: :nilify_all
     end
   end
 
+  test "on delete deleting has_one assoc preserves parent schema prefix" do
+    %MySchema{id: 1}
+    |> Ecto.put_meta(prefix: "prefix")
+    |> Ecto.Changeset.change
+    |> TestRepo.delete!()
+
+    assert_received {:delete_all, %{prefix: "prefix", from: %{source: {"my_assoc", _}}}}
+  end
+
+  test "on delete deleting has_many assocs preserves parent schema prefix" do
+    %MySchema{id: 1}
+    |> Ecto.put_meta(prefix: "prefix")
+    |> Ecto.Changeset.change
+    |> TestRepo.delete!()
+
+    assert_received {:delete_all, %{prefix: "prefix", from: %{source: {"my_assoc", _}}}}
+  end
+
+  test "on delete nilifying has_one assoc preserves parent schema prefix" do
+    %MySchema{id: 1}
+    |> Ecto.put_meta(prefix: "prefix")
+    |> Ecto.Changeset.change
+    |> TestRepo.delete!()
+
+    assert_received {:update_all, %{prefix: "prefix", from: %{source: {"my_assoc", _}}}}
+  end
+
+  test "on delete nilifying has_many assocs preserves parent schema prefix" do
+    %MySchema{id: 1}
+    |> Ecto.put_meta(prefix: "prefix")
+    |> Ecto.Changeset.change
+    |> TestRepo.delete!()
+
+    assert_received {:update_all, %{prefix: "prefix", from: %{source: {"my_assoc", _}}}}
+  end
+
   test "handles assocs on insert" do
     sample = %MyAssoc{x: "xyz"}
 
diff --git a/test/ecto/schema_test.exs b/test/ecto/schema_test.exs
index a5f3eb85..62db0e07 100644
--- a/test/ecto/schema_test.exs
+++ b/test/ecto/schema_test.exs
@@ -422,7 +422,7 @@ defmodule Ecto.SchemaTest do
     end
   end
 
-  test "skipping validations on invalid types" do 
+  test "skipping validations on invalid types" do
     defmodule SchemaSkipValidationsDefault do
       use Ecto.Schema
 
@@ -969,8 +969,18 @@ defmodule Ecto.SchemaTest do
   end
 
   describe "preload_order option" do
+    test "allows MFA" do
+      defmodule MFA do
+        use Ecto.Schema
+
+        schema "assoc" do
+          many_to_many :posts, Post, join_through: "through", preload_order: {__MODULE__, :fun, []}
+        end
+      end
+    end
+
     test "invalid option" do
-      message = "expected `:preload_order` for :posts to be a keyword list or a list of atoms/fields, got: `:title`"
+      message = "expected `:preload_order` for :posts to be a keyword list, a list of atoms/fields or a {Mod, fun, args} tuple, got: `:title`"
       assert_raise ArgumentError, message, fn ->
         defmodule ThroughMatch do
           use Ecto.Schema
diff --git a/test/ecto/type_test.exs b/test/ecto/type_test.exs
index 86da28ff..73344548 100644
--- a/test/ecto/type_test.exs
+++ b/test/ecto/type_test.exs
@@ -216,6 +216,10 @@ defmodule Ecto.TypeTest do
     refute match?(:string, {:param, :any_datetime})
   end
 
+  test "integer" do
+    assert cast(:integer, String.duplicate("1", 64)) == :error
+  end
+
   test "decimal" do
     assert cast(:decimal, "1.0") == {:ok, Decimal.new("1.0")}
     assert cast(:decimal, 1.0) == {:ok, Decimal.new("1.0")}