From 3faf9adb3c84cc61750c9d8df886210e6863f38a Mon Sep 17 00:00:00 2001 From: Dhruv Gupta Date: Fri, 23 Oct 2026 06:25:28 -0400 Subject: [PATCH 1/2] fix: replace duplicated artifact hashes in the artifacts diff When an artifact table holds the same metadata_hash more than once, the stored hashes never equal the newly computed ones again, so every upload reports "Artifacts changed" and the duplicate rows are kept. Delete the rows of duplicated hashes that are still produced and insert those artifacts again once, so a single upload repairs the table. Fixes #1091 --- .../macros/duplicate_table_rows.sql | 12 ++++++++ .../test_dbt_artifacts/test_artifacts.py | 29 +++++++++++++++++++ .../upload_artifacts_to_table.sql | 26 +++++++++++++++++ .../list_utils/get_duplicated_values.sql | 13 +++++++++ 4 files changed, 80 insertions(+) create mode 100644 integration_tests/dbt_project/macros/duplicate_table_rows.sql create mode 100644 macros/utils/list_utils/get_duplicated_values.sql diff --git a/integration_tests/dbt_project/macros/duplicate_table_rows.sql b/integration_tests/dbt_project/macros/duplicate_table_rows.sql new file mode 100644 index 000000000..379e99a59 --- /dev/null +++ b/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 %} diff --git a/integration_tests/tests/test_dbt_artifacts/test_artifacts.py b/integration_tests/tests/test_dbt_artifacts/test_artifacts.py index 4b64fc5f2..1db99e08c 100644 --- a/integration_tests/tests/test_dbt_artifacts/test_artifacts.py +++ b/integration_tests/tests/test_dbt_artifacts/test_artifacts.py @@ -40,6 +40,35 @@ 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." + + def test_artifacts_collection_in_multiple_row_batches(dbt_project: DbtProject): existing_artifacts = dbt_project.read_table("dbt_models") diff --git a/macros/edr/dbt_artifacts/upload_artifacts_to_table.sql b/macros/edr/dbt_artifacts/upload_artifacts_to_table.sql index 12cf9756f..549dc47c9 100644 --- a/macros/edr/dbt_artifacts/upload_artifacts_to_table.sql +++ b/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, diff --git a/macros/utils/list_utils/get_duplicated_values.sql b/macros/utils/list_utils/get_duplicated_values.sql new file mode 100644 index 000000000..99c027b38 --- /dev/null +++ b/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 %} From 47295575a4f1ce78146d8ec6c6caa0dcee11188f Mon Sep 17 00:00:00 2001 From: Dhruv Gupta Date: Tue, 27 Oct 2026 09:51:18 -0400 Subject: [PATCH 2/2] test: check that the repaired artifacts table keeps every hash --- integration_tests/tests/test_dbt_artifacts/test_artifacts.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/integration_tests/tests/test_dbt_artifacts/test_artifacts.py b/integration_tests/tests/test_dbt_artifacts/test_artifacts.py index 1db99e08c..91ff52855 100644 --- a/integration_tests/tests/test_dbt_artifacts/test_artifacts.py +++ b/integration_tests/tests/test_dbt_artifacts/test_artifacts.py @@ -67,6 +67,9 @@ def test_artifacts_diff_removes_duplicated_hashes(dbt_project: DbtProject): 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):