diff --git a/integration_test/myxql/constraints_test.exs b/integration_test/myxql/constraints_test.exs
index 2323823f..15aa2725 100644
--- a/integration_test/myxql/constraints_test.exs
+++ b/integration_test/myxql/constraints_test.exs
@@ -80,4 +80,23 @@ defmodule Ecto.Integration.ConstraintsTest do
       |> PoolRepo.insert()
     assert is_integer(changeset.id)
   end
+
+  @tag :create_constraint
+  test "custom :constraint_handler option" do
+    parent = self()
+    custom_handler = fn _err, _opts ->
+      send(parent, :custom_handler_called)
+      [exclusion: "positive_price"]
+    end
+
+    changeset =
+      %Constraint{}
+      |> Ecto.Changeset.change(price: -10)
+      |> Ecto.Changeset.exclusion_constraint(:price, name: :positive_price)
+
+    {:error, changeset} = PoolRepo.insert(changeset, constraint_handler: custom_handler)
+    assert_received :custom_handler_called
+    assert changeset.errors == [price: {"violates an exclusion constraint", [constraint: :exclusion, constraint_name: "positive_price"]}]
+    assert changeset.data.__meta__.state == :built
+  end
 end
diff --git a/integration_test/pg/constraints_test.exs b/integration_test/pg/constraints_test.exs
index 9e8fe4e4..bc626519 100644
--- a/integration_test/pg/constraints_test.exs
+++ b/integration_test/pg/constraints_test.exs
@@ -112,4 +112,22 @@ defmodule Ecto.Integration.ConstraintsTest do
       |> PoolRepo.insert()
     assert is_integer(changeset.id)
   end
+
+  test "custom :constraint_handler option" do
+    parent = self()
+    custom_handler = fn _err, _opts ->
+      send(parent, :custom_handler_called)
+      [exclusion: "positive_price"]
+    end
+
+    changeset =
+      %Constraint{}
+      |> Ecto.Changeset.change(price: -10)
+      |> Ecto.Changeset.exclusion_constraint(:price, name: :positive_price)
+
+    {:error, changeset} = PoolRepo.insert(changeset, constraint_handler: custom_handler)
+    assert_received :custom_handler_called
+    assert changeset.errors == [price: {"violates an exclusion constraint", [constraint: :exclusion, constraint_name: "positive_price"]}]
+    assert changeset.data.__meta__.state == :built
+  end
 end
diff --git a/lib/ecto/adapters/myxql.ex b/lib/ecto/adapters/myxql.ex
index 6026cb26..2dda844d 100644
--- a/lib/ecto/adapters/myxql.ex
+++ b/lib/ecto/adapters/myxql.ex
@@ -399,7 +399,7 @@ defmodule Ecto.Adapters.MyXQL do
         {:ok, last_insert_id(key, last_insert_id)}
 
       {:error, err} ->
-        case @conn.to_constraints(err, source: source) do
+        case Ecto.Adapters.SQL.to_constraints(adapter_meta, err, opts, source: source) do
           [] -> raise err
           constraints -> {:invalid, constraints}
         end
diff --git a/lib/ecto/adapters/sql.ex b/lib/ecto/adapters/sql.ex
index f7a557c6..658f3568 100644
--- a/lib/ecto/adapters/sql.ex
+++ b/lib/ecto/adapters/sql.ex
@@ -398,6 +398,41 @@ defmodule Ecto.Adapters.SQL do
       {"SELECT p.id, p.title, p.inserted_at, p.created_at FROM posts as p", []}
   """
 
+  @to_constraints_doc """
+  Handles adapter-specific exceptions, converting them to
+  the corresponding constraint errors.
+
+  The constraints are in the keyword list and must return the
+  constraint type, like `:unique`, and the constraint name as
+  a string, for example:
+
+      [unique: "posts_title_index"]
+
+  Returning an empty list signifies the error does not come
+  from any constraint, and should continue with the default
+  exception handling path (i.e. raise or further handling).
+
+  ## Options
+
+    * `:constraint_handler` - a function that receives the exception and
+      error options and returns a keyword list of constraints. Defaults to
+      the adapter connection module's `to_constraints/2`.
+
+      The `:constraint_handler` option can be set per operation or globally
+      via `c:Ecto.Repo.default_options/1` or `c:Ecto.Repo.prepare_options/2`.
+
+  ## Examples
+
+      # Custom handler per operation
+      MyRepo.insert(changeset, constraint_handler: fn
+        %Postgrex.Error{postgres: %{pg_code: "ZZ001", constraint: name}}, _opts ->
+          [check: name]
+
+        err, opts ->
+          Ecto.Adapters.Postgres.Connection.to_constraints(err, opts)
+      end)
+  """
+
   @explain_doc """
   Executes an EXPLAIN statement or similar for the given query according to its kind and the
   adapter in the given repository.
@@ -673,6 +708,28 @@ defmodule Ecto.Adapters.SQL do
     sql_call(adapter_meta, :query_many, [sql], params, opts)
   end
 
+  @doc @to_constraints_doc
+  @spec to_constraints(
+          pid() | Ecto.Repo.t() | Ecto.Adapter.adapter_meta(),
+          exception :: Exception.t(),
+          options :: Keyword.t(),
+          error_options :: Keyword.t()
+        ) :: Keyword.t()
+  def to_constraints(repo, err, opts, err_opts) when is_atom(repo) or is_pid(repo) do
+    to_constraints(Ecto.Adapter.lookup_meta(repo), err, opts, err_opts)
+  end
+
+  def to_constraints(adapter_meta, err, opts, err_opts) do
+    case Keyword.get(opts, :constraint_handler) do
+      handler when is_function(handler, 2) ->
+        handler.(err, err_opts)
+
+      nil ->
+        %{sql: connection} = adapter_meta
+        connection.to_constraints(err, err_opts)
+    end
+  end
+
   defp sql_call(adapter_meta, callback, args, params, opts) do
     %{
       pid: pool,
@@ -1161,7 +1218,7 @@ defmodule Ecto.Adapters.SQL do
   @doc false
   def struct(
         adapter_meta,
-        conn,
+        _conn,
         sql,
         operation,
         source,
@@ -1196,7 +1253,7 @@ defmodule Ecto.Adapters.SQL do
           operation: operation
 
       {:error, err} ->
-        case conn.to_constraints(err, source: source) do
+        case to_constraints(adapter_meta, err, opts, source: source) do
           [] -> raise_sql_call_error(err)
           constraints -> {:invalid, constraints}
         end
diff --git a/test/ecto/adapters/sql_test.exs b/test/ecto/adapters/sql_test.exs
new file mode 100644
index 00000000..0e42bdcd
--- /dev/null
+++ b/test/ecto/adapters/sql_test.exs
@@ -0,0 +1,62 @@
+defmodule Ecto.Adapters.SQLTest.FakeError do
+  defexception [:type, :name, :message]
+end
+
+defmodule Ecto.Adapters.SQLTest.FakeConnection do
+  alias Ecto.Adapters.SQLTest.FakeError
+
+  def to_constraints(%FakeError{type: :unique, name: name}, _opts), do: [unique: name]
+  def to_constraints(%FakeError{type: :check, name: name}, _opts), do: [check: name]
+  def to_constraints(_err, _opts), do: []
+end
+
+defmodule Ecto.Adapters.SQLTest.CustomHandler do
+  alias Ecto.Adapters.SQLTest.{FakeError, FakeConnection}
+
+  def to_constraints(%FakeError{type: :other, name: name}, _opts), do: [exclusion: name]
+  def to_constraints(err, opts), do: FakeConnection.to_constraints(err, opts)
+end
+
+defmodule Ecto.Adapters.SQLTest do
+  use ExUnit.Case, async: true
+
+  alias Ecto.Adapters.SQLTest.{FakeError, FakeConnection, CustomHandler}
+
+  @adapter_meta %{sql: FakeConnection}
+  @unique_err %FakeError{type: :unique, name: "users_email_index", message: "unique violation"}
+  @custom_err %FakeError{type: :other, name: "cannot_overlap", message: "overlap"}
+
+  defp to_constraints(err, opts \\ []) do
+    Ecto.Adapters.SQL.to_constraints(@adapter_meta, err, opts, source: "test")
+  end
+
+  describe "to_constraints/4" do
+    test "uses the adapter connection's to_constraints/2 by default" do
+      assert to_constraints(@unique_err) == [unique: "users_email_index"]
+    end
+
+    test "returns empty list when no constraint matches the default handler" do
+      assert to_constraints(@custom_err) == []
+    end
+
+    test "custom handler handles errors the default handler wouldn't" do
+      assert to_constraints(@custom_err, constraint_handler: &CustomHandler.to_constraints/2) ==
+               [exclusion: "cannot_overlap"]
+    end
+
+    test "custom handler can fall back to the default handler" do
+      assert to_constraints(@unique_err, constraint_handler: &CustomHandler.to_constraints/2) ==
+               [unique: "users_email_index"]
+    end
+
+    test "passes error options to the constraint handler" do
+      handler = fn _err, opts ->
+        send(self(), {:handler_opts, opts})
+        []
+      end
+
+      to_constraints(@custom_err, constraint_handler: handler)
+      assert_received {:handler_opts, [source: "test"]}
+    end
+  end
+end