Add snapshot unique_key hint for duplicate row errors

Author: desusaiteja

.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"

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)

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