diff --git a/target_snowflake/connector.py b/target_snowflake/connector.py
index c8ec2ea..bbffa40 100644
--- a/target_snowflake/connector.py
+++ b/target_snowflake/connector.py
@@ -270,11 +270,13 @@ def _get_merge_from_stage_statement(
         not_matched_insert_values = ", ".join(
             [f's.{col}' for col in formatted_properties]
         )
+        dedup_cols = ", ".join([key for key in formatted_key_properties])
+        dedup = f"QUALIFY ROW_NUMBER() OVER (PARTITION BY {dedup_cols} ORDER BY SEQ8() DESC) = 1"
         return (
             text(
                 f"merge into {full_table_name} d using "
                 + f"(select {', '.join(column_selections)} from '@~/target-snowflake/{sync_id}'"
-                + f"(file_format => {file_format})) s "
+                + f"(file_format => {file_format}) {dedup}) s "
                 + f"on {join_expr} "
                 + f"when matched then update set {matched_clause} "
                 + f"when not matched then insert ({not_matched_insert_cols}) "
diff --git a/tests/test_impl.py b/tests/test_impl.py
index 770058b..7ccdf78 100644
--- a/tests/test_impl.py
+++ b/tests/test_impl.py
@@ -357,8 +357,7 @@ def validate(self) -> None:
         SnowflakeTargetCamelcaseComplexSchema,
         SnowflakeTargetCamelcaseTest,
         TargetCliPrintsTest,
-        # TODO: bug https://github.com/MeltanoLabs/target-snowflake/issues/41
-        # SnowflakeTargetDuplicateRecords,
+        SnowflakeTargetDuplicateRecords,
         SnowflakeTargetEncodedStringData,
         SnowflakeTargetInvalidSchemaTest,
         # Not available in the SDK yet