diff --git a/integration_test/pg/prepare_test.exs b/integration_test/pg/prepare_test.exs
index ea502053..e186446d 100644
--- a/integration_test/pg/prepare_test.exs
+++ b/integration_test/pg/prepare_test.exs
@@ -4,6 +4,28 @@ defmodule Ecto.Integration.PrepareTest do
   alias Ecto.Integration.TestRepo
   alias Ecto.Integration.Post
 
+  # Both sources share their first 51 bytes, so the default statement names
+  # (ecto_insert_<source>_0) would be identical after PostgreSQL's 63-byte
+  # truncation.
+  @long_source "prepare_test_01234567890123456789012345678901234567"
+
+  defmodule LongA do
+    use Ecto.Schema
+
+    schema "prepare_test_01234567890123456789012345678901234567_a" do
+      field :x, :integer
+    end
+  end
+
+  defmodule LongB do
+    use Ecto.Schema
+
+    schema "prepare_test_01234567890123456789012345678901234567_b_logs" do
+      field :x, :integer
+      field :y, :integer
+    end
+  end
+
   test "prepare option" do
     one = TestRepo.insert!(%Post{title: "one"})
     two = TestRepo.insert!(%Post{title: "two"})
@@ -16,4 +38,19 @@ defmodule Ecto.Integration.PrepareTest do
     assert TestRepo.all(Post, prepare: :unnamed) == [one, two]
     assert TestRepo.all(Post, prepare: :named) == [one, two]
   end
+
+  test "statement names longer than 63 bytes do not collide on the server" do
+    TestRepo.query!("CREATE TEMP TABLE #{@long_source}_a (id bigserial PRIMARY KEY, x integer)")
+
+    TestRepo.query!(
+      "CREATE TEMP TABLE #{@long_source}_b_logs (id bigserial PRIMARY KEY, x integer, y integer)"
+    )
+
+    assert %LongA{} = TestRepo.insert!(%LongA{x: 1})
+    assert %LongB{} = TestRepo.insert!(%LongB{x: 1, y: 2})
+    # Without distinct names within 63 bytes this raised 08P01: "bind message
+    # supplies 1 parameters, but prepared statement ... requires 2".
+    assert %LongA{} = TestRepo.insert!(%LongA{x: 3})
+    assert %LongB{} = TestRepo.insert!(%LongB{x: 3, y: 4})
+  end
 end
diff --git a/lib/ecto/adapters/sql.ex b/lib/ecto/adapters/sql.ex
index 4dac125c..caa4b85e 100644
--- a/lib/ecto/adapters/sql.ex
+++ b/lib/ecto/adapters/sql.ex
@@ -987,7 +987,7 @@ defmodule Ecto.Adapters.SQL do
 
     opts =
       if is_nil(Keyword.get(opts, :cache_statement)) do
-        [{:cache_statement, "ecto_insert_all_#{source}"} | opts]
+        [{:cache_statement, cache_statement_name("ecto_insert_all_", source)} | opts]
       else
         opts
       end
@@ -1171,6 +1171,36 @@ defmodule Ecto.Adapters.SQL do
     end
   end
 
+  # PostgreSQL silently truncates prepared statement names to NAMEDATALEN - 1
+  # (63 bytes). Two sources that only differ after that point would share one
+  # server-side statement while the driver caches them under distinct client
+  # names, so the next cached execution binds against the other statement and
+  # fails with "bind message supplies N parameters, but prepared statement
+  # requires M". Keep names within the limit, using a hash to tell them apart.
+  @max_cache_statement_name_size 63
+
+  @doc false
+  def cache_statement_name(prefix, source, suffix \\ "") do
+    source = to_string(source)
+    name = prefix <> source <> suffix
+
+    if byte_size(name) <= @max_cache_statement_name_size do
+      name
+    else
+      hash = source |> :erlang.phash2(4_294_967_296) |> Integer.to_string(36)
+      budget = @max_cache_statement_name_size - byte_size(prefix <> suffix <> hash) - 1
+      kept = source |> truncate_utf8(max(budget, 0)) |> String.trim_trailing("_")
+      prefix <> kept <> "_" <> hash <> suffix
+    end
+  end
+
+  defp truncate_utf8(string, size) when byte_size(string) <= size, do: string
+
+  defp truncate_utf8(string, size) do
+    truncated = binary_part(string, 0, size)
+    if String.valid?(truncated), do: truncated, else: truncate_utf8(string, size - 1)
+  end
+
   @doc false
   def struct(
         adapter_meta,
@@ -1186,7 +1216,8 @@ defmodule Ecto.Adapters.SQL do
       ) do
     opts =
       if is_nil(Keyword.get(opts, :cache_statement)) do
-        [{:cache_statement, "ecto_#{operation}_#{source}_#{length(params)}"} | opts]
+        name = cache_statement_name("ecto_#{operation}_", source, "_#{length(params)}")
+        [{:cache_statement, name} | opts]
       else
         opts
       end
diff --git a/test/ecto/adapters/sql_test.exs b/test/ecto/adapters/sql_test.exs
new file mode 100644
index 00000000..5a8c9ec0
--- /dev/null
+++ b/test/ecto/adapters/sql_test.exs
@@ -0,0 +1,49 @@
+defmodule Ecto.Adapters.SQLTest do
+  use ExUnit.Case, async: true
+
+  import Ecto.Adapters.SQL, only: [cache_statement_name: 2, cache_statement_name: 3]
+
+  describe "cache_statement_name/3" do
+    test "keeps names within the limit untouched" do
+      assert cache_statement_name("ecto_insert_", "posts", "_0") == "ecto_insert_posts_0"
+      assert cache_statement_name("ecto_insert_all_", "posts") == "ecto_insert_all_posts"
+      assert cache_statement_name("ecto_insert_", :posts, "_0") == "ecto_insert_posts_0"
+
+      exactly_63 = String.duplicate("a", 63 - byte_size("ecto_insert__0"))
+      assert byte_size(cache_statement_name("ecto_insert_", exactly_63, "_0")) == 63
+
+      assert cache_statement_name("ecto_insert_", exactly_63, "_0") ==
+               "ecto_insert_#{exactly_63}_0"
+    end
+
+    test "caps long names at 63 bytes while keeping sources distinct" do
+      # These two differ only after PostgreSQL's 63-byte truncation point.
+      a =
+        cache_statement_name(
+          "ecto_insert_",
+          "business_workplace_attendance_leave_comp_rest_minutes",
+          "_0"
+        )
+
+      b =
+        cache_statement_name(
+          "ecto_insert_",
+          "business_workplace_attendance_leave_comp_rest_minutes_event_logs",
+          "_0"
+        )
+
+      assert byte_size(a) <= 63
+      assert byte_size(b) <= 63
+      assert a != b
+      assert a =~ ~r/^ecto_insert_business_workplace_attendance_leave_comp_[0-9A-Z]+_0$/
+      assert String.ends_with?(a, "_0")
+      assert String.ends_with?(b, "_0")
+    end
+
+    test "does not split multibyte characters when truncating" do
+      name = cache_statement_name("ecto_insert_", String.duplicate("é", 40), "_0")
+      assert byte_size(name) <= 63
+      assert String.valid?(name)
+    end
+  end
+end