fix: match the whole key tuple in legacy delete+insert

CHANGELOG.md

@@ -1,5 +1,9 @@
 ## dbt-databricks 1.12.4 (TBD)
 
+### Fixes
+
+- Stop `delete+insert` with a composite `unique_key` from deleting unmatched rows on DBR below 17.1 (thanks @SreeramaYeshwanthGowd!) ([#1612](https://github.com/databricks/dbt-databricks/pull/1612) resolves [#1611](https://github.com/databricks/dbt-databricks/issues/1611))
+
 ### Under the Hood
 
 - Raise the `pytest-rerunfailures` lower bound to `>=16.2` and remove the `SchemaNameVarMixin` workaround so min-deps CI no longer pins 14.0, which leaked class-scoped dbt test fixtures across reruns (test-only, no runtime impact) ([#1618](https://github.com/databricks/dbt-databricks/pull/1618))

dbt/include/databricks/macros/materializations/incremental/strategies.sql

@@ -180,10 +180,19 @@ replace on ({{ replace_on_expr }})
   {%- set statements = [] -%}
 
   {#-- Build WHERE clause for DELETE statement --#}
+  {#-- Match the whole key tuple; per-column IN deletes unmatched combinations (issue #1611) --#}
   {%- set delete_conditions = [] -%}
+  {%- set target_keys = [] -%}
+  {%- set source_keys = [] -%}
   {%- for key in unique_keys -%}
-    {%- do delete_conditions.append(target_relation ~ '.' ~ adapter.quote(key) ~ ' IN (SELECT ' ~ adapter.quote(key) ~ ' FROM ' ~ source_relation ~ ')') -%}
+    {%- do target_keys.append(target_relation ~ '.' ~ adapter.quote(key)) -%}
+    {%- do source_keys.append(adapter.quote(key)) -%}
   {%- endfor -%}
+  {%- if unique_keys | length > 1 -%}
+    {%- do delete_conditions.append('(' ~ target_keys | join(', ') ~ ') IN (SELECT DISTINCT ' ~ source_keys | join(', ') ~ ' FROM ' ~ source_relation ~ ')') -%}
+  {%- else -%}
+    {%- do delete_conditions.append(target_keys[0] ~ ' IN (SELECT ' ~ source_keys[0] ~ ' FROM ' ~ source_relation ~ ')') -%}
+  {%- endif -%}
 
   {#-- Add incremental predicates to DELETE if specified --#}
   {%- if incremental_predicates is sequence and incremental_predicates is not string -%}

tests/functional/adapter/incremental/fixtures.py

@@ -309,6 +309,13 @@
 3,anyway
 """
 
+delete_insert_composite_key_expected = """id,color,msg
+1,blue,hello
+2,red,goodbye
+1,red,updated
+2,blue,updated
+"""
+
 delete_insert_update_schema_expected = """id
 1
 2
@@ -442,6 +449,44 @@
 {% endif %}
 """
 
+force_legacy_delete_insert_macros = """
+{% macro delete_insert_sql_impl(
+     source_relation, target_relation, target_columns, unique_key, incremental_predicates
+   ) %}
+  {#-- Force the DBR < 17.1 path so the legacy DELETE predicate runs on any compute --#}
+  {%- set keys = unique_key
+        if unique_key is sequence and unique_key is not string
+        else [unique_key] -%}
+  {% do return(delete_insert_legacy_sql(
+       source_relation, target_relation, target_columns, keys, incremental_predicates
+     )) %}
+{% endmacro %}
+"""
+
+delete_insert_composite_key_model = """
+{{ config(
+    materialized = 'incremental',
+    unique_key = ['id', 'color'],
+    incremental_strategy = 'delete+insert',
+) }}
+
+{% if not is_incremental() %}
+
+select cast(1 as bigint) as id, 'blue' as color, 'hello' as msg
+union all
+select cast(2 as bigint) as id, 'red' as color, 'goodbye' as msg
+
+{% else %}
+
+-- Neither key tuple exists in the target, so nothing should be deleted.
+-- Matching each key column on its own would delete both existing rows.
+select cast(1 as bigint) as id, 'red' as color, 'updated' as msg
+union all
+select cast(2 as bigint) as id, 'blue' as color, 'updated' as msg
+
+{% endif %}
+"""
+
 delete_insert_with_predicates_model = """
 {{ config(
     materialized = 'incremental',

tests/functional/adapter/incremental/test_incremental_strategies.py

@@ -451,6 +451,30 @@ def test_incremental(self, project):
         )
 
 
+class TestDeleteInsertCompositeKey(IncrementalBase):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "delete_insert_model.sql": fixtures.delete_insert_composite_key_model,
+        }
+
+    @pytest.fixture(scope="class")
+    def macros(self):
+        return {"force_legacy_delete_insert.sql": fixtures.force_legacy_delete_insert_macros}
+
+    @pytest.fixture(scope="class")
+    def seeds(self):
+        return {
+            "delete_insert_expected.csv": fixtures.delete_insert_composite_key_expected,
+        }
+
+    def test_incremental(self, project):
+        self.seed_and_run_twice()
+        util.check_relations_equal(
+            project.adapter, ["delete_insert_model", "delete_insert_expected"]
+        )
+
+
 class TestDeleteInsertUpdateSchema(IncrementalBase):
     @pytest.fixture(scope="class")
     def models(self):

tests/unit/macros/materializations/incremental/test_delete_insert.py

@@ -153,44 +153,35 @@ def test_delete_insert_legacy_sql__non_ascii_unique_key(self, template, context)
         assert self.clean_sql(insert_sql).startswith("insert into target")
 
     def test_delete_insert_legacy_sql__multiple_unique_keys(self, template, context):
-        """Multiple unique keys are each back-quoted and ANDed in the DELETE predicate."""
+        """A composite unique_key matches the whole tuple, not each column independently."""
         delete_sql, _ = self.render_legacy(
             template, context, unique_keys=["a", "b"], target_columns=("a", "b")
         )
         clean_delete = self.clean_sql(delete_sql)
-        assert "target.`a` in (select `a` from source)" in clean_delete
-        assert "target.`b` in (select `b` from source)" in clean_delete
-        assert clean_delete.count(" and ") == 1
-
-    def test_legacy_sql_generation__single_unique_key_delete(self, template, context):
-        """Test the DELETE SQL generation for single unique key"""
-        # We'll verify by compiling a test query that uses the same logic
-        # Mock adapter
-        context["adapter"].has_dbr_capability = lambda cap: cap == "insert_by_name"
-
-        # Build expected DELETE manually using the same logic as the macro
-        expected_delete = """
-            delete from target
-            where target.a IN (SELECT a FROM source)
-            """
-
-        # The macro builds: target.{key} IN (SELECT {key} FROM source)
-        # This test documents the expected SQL pattern
-        assert "delete from" in expected_delete.lower()
-        assert "target.a in (select a from source)" in expected_delete.lower()
+        assert "(target.`a`, target.`b`) in (select distinct `a`, `b` from source)" in clean_delete
+        assert clean_delete.count(" and ") == 0
 
-    def test_legacy_sql_generation__multiple_unique_keys_delete(self, template, context):
-        """Test the DELETE SQL generation for multiple unique keys"""
-        expected_delete = """
-            delete from target
-            where target.a IN (SELECT a FROM source)
-              and target.b IN (SELECT b FROM source)
-            """
+    def test_delete_insert_legacy_sql__multiple_unique_keys_with_predicates(
+        self, template, context
+    ):
+        """Incremental predicates are ANDed after the tuple match, not inside it."""
+        delete_sql, _ = self.render_legacy(
+            template,
+            context,
+            unique_keys=["a", "b"],
+            target_columns=("a", "b"),
+            incremental_predicates=["a > 1"],
+        )
+        clean_delete = self.clean_sql(delete_sql)
+        assert "(target.`a`, target.`b`) in (select distinct `a`, `b` from source)" in clean_delete
+        assert "and a > 1" in clean_delete
 
-        # The macro builds conditions for each key with AND
-        assert "target.a in" in expected_delete.lower()
-        assert "target.b in" in expected_delete.lower()
-        assert expected_delete.lower().count(" and ") == 1
+    def test_legacy_sql_generation__single_unique_key_delete(self, template, context):
+        """A single unique key keeps the original per-column predicate."""
+        delete_sql, _ = self.render_legacy(template, context, unique_keys=["a"])
+        clean_delete = self.clean_sql(delete_sql)
+        assert clean_delete.startswith("delete from target where")
+        assert "target.`a` in (select `a` from source)" in clean_delete
 
     def test_legacy_sql_generation__with_predicates_delete(self, template, context):
         """Test that incremental_predicates are added to DELETE WHERE clause"""