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..91ff52855 100644 --- a/integration_tests/tests/test_dbt_artifacts/test_artifacts.py +++ b/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") 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 %}