From f23c06cd57abbb7ad891f3e1b6b009dfc6258e91 Mon Sep 17 00:00:00 2001
From: Sai Teja Desu <desu.saiteja@gmail.com>
Date: Mon, 18 May 2026 20:20:15 -0700
Subject: [PATCH] Add snapshot unique_key hint for duplicate row errors

---
 .../unreleased/Fixes-20260518-075256.yaml     |  6 +++
 core/dbt/task/snapshot.py                     | 17 +++++++
 tests/unit/task/test_snapshot.py              | 44 +++++++++++++++++++
 3 files changed, 67 insertions(+)
 create mode 100644 .changes/unreleased/Fixes-20260518-075256.yaml
 create mode 100644 tests/unit/task/test_snapshot.py

diff --git a/.changes/unreleased/Fixes-20260518-075256.yaml b/.changes/unreleased/Fixes-20260518-075256.yaml
new file mode 100644
index 00000000000..a3f10fb9143
--- /dev/null
+++ b/.changes/unreleased/Fixes-20260518-075256.yaml
@@ -0,0 +1,6 @@
+kind: Fixes
+body: "Add a snapshot-specific hint when duplicate rows during DML suggest a non-unique `unique_key`"
+time: 2026-05-18T07:52:56.000000-07:00
+custom:
+    Author: saitejadesu
+    Issue: "10864"
diff --git a/core/dbt/task/snapshot.py b/core/dbt/task/snapshot.py
index 37ef62218be..b7ffc25f5fc 100644
--- a/core/dbt/task/snapshot.py
+++ b/core/dbt/task/snapshot.py
@@ -13,10 +13,27 @@
 from dbt_common.utils import cast_dict_to_dict_of_strings
 
 
+_DUPLICATE_ROW_DML_ERROR = "duplicate row detected during dml action"
+SNAPSHOT_UNIQUE_KEY_HINT = (
+    "Hint: This can happen when the snapshot's configured `unique_key` is not unique. "
+    "Ensure the `unique_key` column(s) identify a single row in the snapshot query."
+)
+
+
+def _add_snapshot_unique_key_hint(message: str) -> str:
+    if _DUPLICATE_ROW_DML_ERROR not in message.lower() or SNAPSHOT_UNIQUE_KEY_HINT in message:
+        return message
+    return f"{message}\n\n{SNAPSHOT_UNIQUE_KEY_HINT}"
+
+
 class SnapshotRunner(ModelRunner):
     def describe_node(self) -> str:
         return "snapshot {}".format(self.get_node_representation())
 
+    def handle_exception(self, e: Exception, ctx) -> str:
+        message = super().handle_exception(e, ctx)
+        return _add_snapshot_unique_key_hint(message)
+
     def print_result_line(self, result):
         model = result.node
         group = group_lookup.get(model.unique_id)
diff --git a/tests/unit/task/test_snapshot.py b/tests/unit/task/test_snapshot.py
new file mode 100644
index 00000000000..d94858fc391
--- /dev/null
+++ b/tests/unit/task/test_snapshot.py
@@ -0,0 +1,44 @@
+from unittest.mock import patch
+
+from dbt.task.run import ModelRunner
+from dbt.task.snapshot import (
+    SNAPSHOT_UNIQUE_KEY_HINT,
+    SnapshotRunner,
+    _add_snapshot_unique_key_hint,
+)
+
+
+def test_add_snapshot_unique_key_hint_for_duplicate_row_dml_error():
+    message = "Database Error\n  Duplicate row detected during DML action"
+
+    updated_message = _add_snapshot_unique_key_hint(message)
+
+    assert message in updated_message
+    assert SNAPSHOT_UNIQUE_KEY_HINT in updated_message
+
+
+def test_add_snapshot_unique_key_hint_only_once():
+    message = (
+        "Database Error\n"
+        "  Duplicate row detected during DML action\n\n"
+        f"{SNAPSHOT_UNIQUE_KEY_HINT}"
+    )
+
+    assert _add_snapshot_unique_key_hint(message) == message
+
+
+def test_add_snapshot_unique_key_hint_ignores_other_errors():
+    message = "Database Error\n  relation does not exist"
+
+    assert _add_snapshot_unique_key_hint(message) == message
+
+
+def test_snapshot_runner_adds_unique_key_hint_to_duplicate_row_error():
+    message = "Database Error\n  Duplicate row detected during DML action"
+    runner = object.__new__(SnapshotRunner)
+
+    with patch.object(ModelRunner, "handle_exception", return_value=message):
+        updated_message = runner.handle_exception(Exception("database failed"), object())
+
+    assert message in updated_message
+    assert SNAPSHOT_UNIQUE_KEY_HINT in updated_message