diff --git a/lib/ecto/query.ex b/lib/ecto/query.ex
index 7ba6197f8c..057825b419 100644
--- a/lib/ecto/query.ex
+++ b/lib/ecto/query.ex
@@ -435,6 +435,11 @@ defmodule Ecto.Query do
     defstruct [recursive: false, queries: []]
   end
 
+  defmodule LimitExpr do
+    @moduledoc false
+    defstruct [:expr, :file, :line, :with_ties, params: []]
+  end
+
   defmodule Tagged do
     @moduledoc false
     # * value is the tagged value
@@ -961,6 +966,8 @@ defmodule Ecto.Query do
         end
       end
 
+    {t, quoted} = maybe_with_ties(type, t, quoted, binds)
+
     from(t, env, count_bind, quoted, binds)
   end
 
@@ -983,6 +990,10 @@ defmodule Ecto.Query do
     from(t, env, count_bind, quoted, to_query_binds(binds))
   end
 
+  defp from([{:with_ties, _value}|_], _env, _count_bind, _quoted, _binds) do
+    Builder.error! "`with_ties` keyword must immediately follow a limit"
+  end
+
   defp from([{:on, _value}|_], _env, _count_bind, _quoted, _binds) do
     Builder.error! "`on` keyword must immediately follow a join"
   end
@@ -999,6 +1010,23 @@ defmodule Ecto.Query do
     quoted
   end
 
+  defp maybe_with_ties(:limit, t, quoted, binds) do
+    {t, with_ties} = collect_with_ties(t, nil)
+
+    quoted =
+      if with_ties != nil do
+        quote do
+          Ecto.Query.with_ties(unquote(quoted), unquote(binds), unquote(with_ties))
+        end
+      else
+        quoted
+      end
+
+    {t, quoted}
+  end
+
+  defp maybe_with_ties(_type, t, quoted, _binds), do: {t, quoted}
+
   defp to_query_binds(binds) do
     for {k, v} <- binds, do: {{k, [], nil}, v}
   end
@@ -1013,6 +1041,13 @@ defmodule Ecto.Query do
   defp join_qual(:left_lateral_join), do: :left_lateral
   defp join_qual(:inner_lateral_join), do: :inner_lateral
 
+  defp collect_with_ties([{:with_ties, with_ties} | t], nil),
+    do: collect_with_ties(t, with_ties)
+  defp collect_with_ties([{:with_ties, _} | _], _),
+    do: Builder.error! "`with_ties` keyword was given more than once to the same limit"
+  defp collect_with_ties(t, with_ties),
+    do: {t, with_ties}
+
   defp collect_on([{key, _} | _] = t, on, as, prefix, hints) when key in @from_join_opts do
     {t, as, prefix, hints} = collect_as_and_prefix_and_hints(t, as, prefix, hints)
     collect_on(t, on, as, prefix, hints)
@@ -1955,6 +1990,34 @@ defmodule Ecto.Query do
     Builder.LimitOffset.build(:limit, query, binding, expr, __CALLER__)
   end
 
+  @doc """
+  Enables or disables ties for limit expressions.
+
+  If there are multiple records tied for the last position in an ordered
+  limit result, setting this value to `true` will return all of the tied
+  records, even if the final result exceeds the specified limit.
+
+  Must be a boolean or evaluate to a boolean at runtime. Can only be applied
+  to queries with a `limit` expression or an error is raised. If `limit`
+  is redefined then `with_ties` must be reapplied.
+
+  Not all databases support this option and the ones that do might list it
+  under the `FETCH` command. Databases may require a corresponding `order_by`
+  statement to evaluate ties.
+
+  ## Keywords example
+
+      from(p in Post, where: p.author_id == ^current_user, order_by: [desc: p.visits], limit: 10, with_ties: true)
+
+  ## Expressions example
+
+      Post |> where([p], p.author_id == ^current_user) |> order_by([p], desc: p.visits) |> limit(10) |> with_ties(true)
+
+  """
+  defmacro with_ties(query, binding \\ [], expr) do
+    Builder.LimitOffset.build(:with_ties, query, binding, expr, __CALLER__)
+  end
+
   @doc """
   An offset query expression.
 
@@ -2353,7 +2416,7 @@ defmodule Ecto.Query do
   def last(queryable, key), do: last(order_by(queryable, ^key), nil)
 
   defp limit do
-    %QueryExpr{expr: 1, params: [], file: __ENV__.file, line: __ENV__.line}
+    %LimitExpr{expr: 1, params: [], file: __ENV__.file, line: __ENV__.line}
   end
 
   defp field(ix, field) when is_integer(ix) and is_atom(field) do
diff --git a/lib/ecto/query/builder/limit_offset.ex b/lib/ecto/query/builder/limit_offset.ex
index 8464afb5c8..62f18b2779 100644
--- a/lib/ecto/query/builder/limit_offset.ex
+++ b/lib/ecto/query/builder/limit_offset.ex
@@ -5,6 +5,15 @@ defmodule Ecto.Query.Builder.LimitOffset do
 
   alias Ecto.Query.Builder
 
+  @doc """
+  Validates `with_ties` at runtime.
+  """
+  @spec with_ties!(any) :: boolean
+  def with_ties!(with_ties) when is_boolean(with_ties), do: with_ties
+
+  def with_ties!(with_ties),
+    do: raise("`with_ties` expression must evaluate to a boolean at runtime, got: `#{inspect(with_ties)}`")
+
   @doc """
   Builds a quoted expression.
 
@@ -12,32 +21,92 @@ defmodule Ecto.Query.Builder.LimitOffset do
   If possible, it does all calculations at compile time to avoid
   runtime work.
   """
-  @spec build(:limit | :offset, Macro.t, [Macro.t], Macro.t, Macro.Env.t) :: Macro.t
+  @spec build(:limit | :with_ties | :offset, Macro.t(), [Macro.t()], Macro.t(), Macro.Env.t()) ::
+          Macro.t()
   def build(type, query, binding, expr, env) do
-    {query, binding} = Builder.escape_binding(query, binding, env)
-    {expr, {params, _acc}} = Builder.escape(expr, :integer, {[], %{}}, binding, env)
+    {query, vars} = Builder.escape_binding(query, binding, env)
+    {expr, {params, _acc}} = escape(type, expr, {[], %{}}, vars, env)
     params = Builder.escape_params(params)
+    quoted = build_quoted(type, expr, params, env)
+
+    Builder.apply_query(query, __MODULE__, [type, quoted], env)
+  end
+
+  defp escape(type, expr, params_acc, vars, env) when type in [:limit, :offset] do
+    Builder.escape(expr, :integer, params_acc, vars, env)
+  end
+
+  defp escape(:with_ties, expr, params_acc, _vars, _env) when is_boolean(expr) do
+    {expr, params_acc}
+  end
+
+  defp escape(:with_ties, {:^, _, [expr]}, params_acc, _vars, _env) do
+    {quote(do: Ecto.Query.Builder.LimitOffset.with_ties!(unquote(expr))), params_acc}
+  end
+
+  defp escape(:with_ties, expr, _params_acc, _vars, _env) do
+    Builder.error!(
+      "`with_ties` expression must be a compile time boolean or an interpolated value using ^, got: `#{Macro.to_string(expr)}`"
+    )
+  end
 
-    limoff = quote do: %Ecto.Query.QueryExpr{
-                        expr: unquote(expr),
-                        params: unquote(params),
-                        file: unquote(env.file),
-                        line: unquote(env.line)}
+  defp build_quoted(:limit, expr, params, env) do
+    quote do: %Ecto.Query.LimitExpr{
+            expr: unquote(expr),
+            params: unquote(params),
+            with_ties: false,
+            file: unquote(env.file),
+            line: unquote(env.line)
+          }
+  end
 
-    Builder.apply_query(query, __MODULE__, [type, limoff], env)
+  defp build_quoted(:offset, expr, params, env) do
+    quote do: %Ecto.Query.QueryExpr{
+            expr: unquote(expr),
+            params: unquote(params),
+            file: unquote(env.file),
+            line: unquote(env.line)
+          }
   end
 
+  defp build_quoted(:with_ties, expr, _params, _env), do: expr
+
   @doc """
   The callback applied by `build/4` to build the query.
   """
-  @spec apply(Ecto.Queryable.t, :limit | :offset, term) :: Ecto.Query.t
+  @spec apply(Ecto.Queryable.t(), :limit | :with_ties | :offset, term) :: Ecto.Query.t()
   def apply(%Ecto.Query{} = query, :limit, expr) do
     %{query | limit: expr}
   end
+
+  def apply(%Ecto.Query{limit: limit} = query, :with_ties, expr) do
+    %{query | limit: apply_limit(limit, expr)}
+  end
+
   def apply(%Ecto.Query{} = query, :offset, expr) do
     %{query | offset: expr}
   end
+
   def apply(query, kind, expr) do
     apply(Ecto.Queryable.to_query(query), kind, expr)
   end
+
+  @doc """
+  Applies the `with_ties` value to the `limit` struct.
+  """
+  def apply_limit(nil, _with_ties) do
+    Builder.error!("`with_ties` can only be applied to queries containing a `limit`")
+  end
+
+  # Runtime
+  def apply_limit(%_{} = limit, with_ties) do
+    %{limit | with_ties: with_ties}
+  end
+
+  # Compile
+  def apply_limit(limit, with_ties) do
+    quote do
+      Ecto.Query.Builder.LimitOffset.apply_limit(unquote(limit), unquote(with_ties))
+    end
+  end
 end
diff --git a/lib/ecto/query/inspect.ex b/lib/ecto/query/inspect.ex
index 14d07d10b1..4d4813e5c1 100644
--- a/lib/ecto/query/inspect.ex
+++ b/lib/ecto/query/inspect.ex
@@ -1,7 +1,7 @@
 import Inspect.Algebra
 import Kernel, except: [to_string: 1]
 
-alias Ecto.Query.{DynamicExpr, JoinExpr, QueryExpr, WithExpr}
+alias Ecto.Query.{DynamicExpr, JoinExpr, QueryExpr, WithExpr, LimitExpr}
 
 defimpl Inspect, for: Ecto.Query.DynamicExpr do
   def inspect(%DynamicExpr{binding: binding} = dynamic, opts) do
@@ -95,6 +95,7 @@ defimpl Inspect, for: Ecto.Query do
     assocs = assocs(query.assocs, names)
     windows = windows(query.windows, names)
     combinations = combinations(query.combinations)
+    limit = limit(query.limit, names)
 
     wheres = bool_exprs(%{and: :where, or: :or_where}, query.wheres, names)
     group_bys = kw_exprs(:group_by, query.group_bys, names)
@@ -103,7 +104,6 @@ defimpl Inspect, for: Ecto.Query do
     updates = kw_exprs(:update, query.updates, names)
 
     lock = kw_inspect(:lock, query.lock)
-    limit = kw_expr(:limit, query.limit, names)
     offset = kw_expr(:offset, query.offset, names)
     select = kw_expr(:select, query.select, names)
     distinct = kw_expr(:distinct, query.distinct, names)
@@ -191,6 +191,16 @@ defimpl Inspect, for: Ecto.Query do
     Enum.map(combinations, fn {key, val} -> {key, "(" <> to_string(val) <> ")"} end)
   end
 
+  defp limit(nil, _names), do: []
+
+  defp limit(%LimitExpr{with_ties: false} = limit, names) do
+    [{:limit, expr(limit, names)}]
+  end
+
+  defp limit(%LimitExpr{with_ties: with_ties} = limit, names) do
+    [{:limit, expr(limit, names)}] ++ kw_inspect(:with_ties, with_ties)
+  end
+
   defp bool_exprs(keys, exprs, names) do
     Enum.map(exprs, fn %{expr: expr, op: op} = part ->
       {Map.fetch!(keys, op), expr(expr, names, part)}
diff --git a/lib/ecto/query/planner.ex b/lib/ecto/query/planner.ex
index 0fa9374f46..4bdad40c30 100644
--- a/lib/ecto/query/planner.ex
+++ b/lib/ecto/query/planner.ex
@@ -2,7 +2,7 @@ defmodule Ecto.Query.Planner do
   # Normalizes a query and its parameters.
   @moduledoc false
 
-  alias Ecto.Query.{BooleanExpr, DynamicExpr, FromExpr, JoinExpr, QueryExpr, SelectExpr}
+  alias Ecto.Query.{BooleanExpr, DynamicExpr, FromExpr, JoinExpr, QueryExpr, SelectExpr, LimitExpr}
 
   if map_size(%Ecto.Query{}) != 21 do
     raise "Ecto.Query match out of date in builder"
@@ -773,6 +773,7 @@ defmodule Ecto.Query.Planner do
     # Current strategy appends [{:subquery, i, cache}], where cache is the cache key for this subquery.
     {op, expr, Enum.map(subqueries, fn %{cache: cache} -> {:subquery, cache} end)}
   end
+  defp expr_to_cache(%LimitExpr{expr: expr, with_ties: with_ties}), do: {with_ties, expr}
 
   @spec cast_and_merge_params(atom, Ecto.Query.t, any, list, module) :: {params :: list, cacheable? :: boolean}
   defp cast_and_merge_params(kind, query, expr, params, adapter) do
diff --git a/test/ecto/query/builder/limit_offset_test.exs b/test/ecto/query/builder/limit_offset_test.exs
index 767b48972a..6b8b537737 100644
--- a/test/ecto/query/builder/limit_offset_test.exs
+++ b/test/ecto/query/builder/limit_offset_test.exs
@@ -1,7 +1,18 @@
+Code.require_file "../../../support/eval_helpers.exs", __DIR__
+
 defmodule Ecto.Query.Builder.LimitOffsetTest do
   use ExUnit.Case, async: true
 
   import Ecto.Query
+  import Support.EvalHelpers
+
+  defmodule Post do
+    use Ecto.Schema
+
+    schema "posts" do
+      field :title
+    end
+  end
 
   test "overrides on duplicated limit and offset" do
     query = "posts" |> limit([], 1) |> limit([], 2)
@@ -10,4 +21,47 @@ defmodule Ecto.Query.Builder.LimitOffsetTest do
     query = "posts" |> offset([], 1) |> offset([], 2) |> select([], 3)
     assert query.offset.expr == 2
   end
+
+  test "with_ties" do
+    # compile time query
+    query = from(p in Post) |> limit([], 1) |> with_ties(true)
+    assert query.limit.expr == 1
+    assert query.limit.with_ties == true
+
+    # runtime query
+    query = "posts" |> limit([], 2) |> with_ties(^true)
+    assert query.limit.expr == 2
+    assert query.limit.with_ties == true
+  end
+
+  test "with_ties is removed when a new limit is set" do
+    # compile time query
+    query = from(p in Post) |> limit([], 1) |> with_ties(true) |> limit([], 2)
+    assert query.limit.expr == 2
+    assert query.limit.with_ties == false
+
+    # runtime query
+    query = "posts" |> limit([], 3) |> with_ties(true) |> limit([], 4)
+    assert query.limit.expr == 4
+    assert query.limit.with_ties == false
+  end
+
+  test "with_ties must be a runtime or compile time boolean" do
+    msg = "`with_ties` expression must evaluate to a boolean at runtime, got: `1`"
+    assert_raise RuntimeError, msg, fn ->
+      with_ties("posts", ^1)
+    end
+
+    msg = "`with_ties` expression must be a compile time boolean or an interpolated value using ^, got: `1`"
+    assert_raise Ecto.Query.CompileError, msg, fn ->
+      quote_and_eval with_ties("posts", 1)
+    end
+  end
+
+  test "with_ties requires a limit" do
+    msg = "`with_ties` can only be applied to queries containing a `limit`"
+    assert_raise Ecto.Query.CompileError, msg, fn ->
+      with_ties("posts", true)
+    end
+  end
 end
diff --git a/test/ecto/query/planner_test.exs b/test/ecto/query/planner_test.exs
index ebeebb9c36..1b086dcb5b 100644
--- a/test/ecto/query/planner_test.exs
+++ b/test/ecto/query/planner_test.exs
@@ -476,6 +476,7 @@ defmodule Ecto.Query.PlannerTest do
         on: true,
         hints: ["join hint"],
         prefix: "world",
+        limit: 1, with_ties: true,
         preload: :comments
       )
 
@@ -483,6 +484,7 @@ defmodule Ecto.Query.PlannerTest do
     assert key == [:all,
                    {:lock, "foo"},
                    {:prefix, "foo"},
+                   {:limit, {true, 1}},
                    {:where, [{:and, {:is_nil, [], [nil]}}, {:or, {:is_nil, [], [nil]}}]},
                    {:join, [{:inner, {"comments", Comment, 38292156, "world"}, true, ["join hint"]}]},
                    {:from, {"posts", Post, 32915161, "hello"}, ["hint"]},
@@ -749,7 +751,7 @@ defmodule Ecto.Query.PlannerTest do
                :all,
                {:prefix, "another"},
                {:take, %{0 => {:any, [:id]}}},
-               {:limit, {:^, [], [0]}},
+               {:limit, {false, {:^, [], [0]}}},
                {:order_by, [[desc: _]]},
                {:from, {"comments", Comment, _, nil}, []},
                {:select, {:&, [], [0]}}
@@ -781,7 +783,7 @@ defmodule Ecto.Query.PlannerTest do
                :all,
                {:prefix, "another"},
                {:take, %{0 => {:any, [:id]}}},
-               {:limit, {:^, [], [0]}},
+               {:limit, {false, {:^, [], [0]}}},
                {:order_by, [[desc: _]]},
                {:from, {"comments", Comment, _, nil}, []},
                {:select, {:&, [], [0]}}