fix: replace duplicated artifact hashes in the artifacts diff

integration_tests/dbt_project/macros/duplicate_table_rows.sql

@@ -0,0 +1,12 @@
+{% macro duplicate_table_rows(table_name) %}
+    {#- Insert a copy of every row of the given Elementary table, e.g. to
+        simulate two overlapping artifact uploads to the same schema. -#}
+    {% set relation = ref(table_name) %}
+    {% do run_query("INSERT INTO " ~ relation ~ " SELECT * FROM " ~ relation) %}
+
+    {#- Same commit handling as insert_sentinel_row. -#}
+    {% set no_commit_adapters = ["spark", "databricks", "bigquery", "athena"] %}
+    {% if target.type not in no_commit_adapters %}
+        {% do run_query("COMMIT") %}
+    {% endif %}
+{% endmacro %}

integration_tests/tests/test_dbt_artifacts/test_artifacts.py

@@ -40,6 +40,38 @@ def test_artifacts_updating(dbt_project: DbtProject):
     assert first_row != second_row, "Artifacts are not updated at the on-run-end."
 
 
+@pytest.mark.skip_targets(["dremio"])
+def test_artifacts_diff_removes_duplicated_hashes(dbt_project: DbtProject):
+    # Disabled by default in the tests for performance reasons.
+    dbt_project.dbt_runner.vars["disable_dbt_artifacts_autoupload"] = False
+
+    dbt_project.dbt_runner.vars["cache_artifacts"] = True
+
+    dbt_project.dbt_runner.run(select=TEST_MODEL)
+    metadata_hashes = [
+        row["metadata_hash"] for row in dbt_project.read_table("dbt_models")
+    ]
+    assert len(metadata_hashes) == len(set(metadata_hashes))
+
+    # Leave every stored hash duplicated, as two overlapping uploads would.
+    dbt_project.dbt_runner.run_operation(
+        "elementary_tests.duplicate_table_rows",
+        macro_args={"table_name": "dbt_models"},
+    )
+    assert len(dbt_project.read_table("dbt_models")) == 2 * len(metadata_hashes)
+
+    dbt_project.dbt_runner.run(select=TEST_MODEL)
+    new_metadata_hashes = [
+        row["metadata_hash"] for row in dbt_project.read_table("dbt_models")
+    ]
+    assert len(new_metadata_hashes) == len(
+        set(new_metadata_hashes)
+    ), "Duplicated artifacts were not removed by the diff upload."
+    assert set(new_metadata_hashes) == set(
+        metadata_hashes
+    ), "Artifacts were lost while removing the duplicated ones."
+
+
 def test_artifacts_collection_in_multiple_row_batches(dbt_project: DbtProject):
     existing_artifacts = dbt_project.read_table("dbt_models")
 

macros/edr/dbt_artifacts/upload_artifacts_to_table.sql

@@ -76,6 +76,32 @@
                     | reject("in", new_metadata_hashes)
                     | list
                 ) %}
+                {#- A hash stored more than once (e.g. after two overlapping uploads)
+                    keeps the stored hashes from ever matching the new ones, so its
+                    rows are deleted and the artifact is inserted again once. -#}
+                {% set duplicated_artifact_hashes = (
+                    elementary.get_duplicated_values(
+                        upload_artifact_method.metadata_hashes
+                    )
+                    | select("in", new_metadata_hashes)
+                    | list
+                ) %}
+                {% if duplicated_artifact_hashes %}
+                    {% do elementary.file_log(
+                        "[{}] Replacing {} duplicated artifacts.".format(
+                            table_relation.identifier,
+                            duplicated_artifact_hashes | length,
+                        )
+                    ) %}
+                    {% do added_artifacts.extend(
+                        flatten_artifact_dicts
+                        | selectattr(
+                            "metadata_hash", "in", duplicated_artifact_hashes
+                        )
+                        | list
+                    ) %}
+                    {% do removed_artifact_hashes.extend(duplicated_artifact_hashes) %}
+                {% endif %}
                 {% do elementary.delete_and_insert(
                     table_relation,
                     insert_rows=added_artifacts,

macros/utils/list_utils/get_duplicated_values.sql

@@ -0,0 +1,13 @@
+{% macro get_duplicated_values(values) %}
+    {#- Returns the values that appear more than once, each of them once. -#}
+    {% set seen_values = {} %}
+    {% set duplicated_values = [] %}
+    {% for value in values %}
+        {% if value not in seen_values %} {% do seen_values.update({value: false}) %}
+        {% elif not seen_values[value] %}
+            {% do seen_values.update({value: true}) %}
+            {% do duplicated_values.append(value) %}
+        {% endif %}
+    {% endfor %}
+    {% do return(duplicated_values) %}
+{% endmacro %}