fix: match the whole key tuple in legacy delete+insert (#1612)

Resolves #1611

### Description

On the `delete+insert` strategy with a composite `unique_key`, the
legacy DBR < 17.1 path builds one
`IN (SELECT ...)` per key column and ANDs them, so the DELETE predicate
is the cross product of the
key columns' value sets rather than the set of key tuples. Rows whose
key tuple is absent from the
source are deleted and never re-inserted, silently.

With `unique_key: ['a', 'b']`, a target holding `(1,10) (2,20) (1,20)
(2,10)` and a source producing
only `(1,10) (2,20)`, the predicate `a IN (1,2) AND b IN (10,20)`
matches all four rows. `(1,20)` and
`(2,10)` are lost.

The DBR 17.1+ branch of the same macro is already row-wise, and
dbt-core's cross-adapter default is
`where (unique_key_str) in (select distinct unique_key_str from
source)`, so today the same model with
the same config produces different data depending on runtime version.

**Fix:** use a correlated `EXISTS` predicate that compares every key
column against the same source
row with null-safe equality.

```sql
delete from target
where exists (
  select 1
  from source
  where target.`a` <=> source.`a`
    and target.`b` <=> source.`b`
)
```

The single-key branch remains byte-for-byte identical, so its behavior
is unchanged and remains
covered by the existing single-key and non-ASCII tests.

**Note on the changed tests:**
`test_delete_insert_legacy_sql__multiple_unique_keys` previously
asserted the per-column form, so it encoded the bug and had to be
updated. Two neighbouring tests that
asserted against their own hardcoded strings are converted to real macro
renders, following your
request on #1595.

**Testing:**

- 17 focused unit tests cover composite keys, incremental predicates,
and the unchanged single-key
  path.
- A functional regression forces the legacy `DELETE` + `INSERT` branch
on live Databricks and
verifies exact replacements, crossed-tuple preservation, and absent-row
preservation.
- The correlated `DELETE` statement was also run directly on DBR 16.4
LTS and produced the expected
  final rows.

A full dbt end-to-end run on compute below DBR 17.1 would add
release-matrix confidence, but the
changed predicate and complete dbt execution path are covered by the
complementary checks above.

### Checklist

- [x] I have run this code in development and it appears to resolve the
stated issue
- [x] This PR includes tests, or tests are not required/relevant for
this PR
- [x] I have updated the `CHANGELOG.md` and added information about my
change to the "dbt-databricks next" section.

---------

Signed-off-by: Sreerama Yeshwanth Gowd <yeshwanthgowdsreerama@gmail.com>
Co-authored-by: Shubham Dhal <shubham.dhal@databricks.com>

Author: SreeramaYeshwanthGowd

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ### 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))
 - Escape single quotes in relation comments so materialized views and streaming tables with an apostrophe in the description can be created (thanks @SreeramaYeshwanthGowd!) ([#1613](https://github.com/databricks/dbt-databricks/pull/1613) resolves [#1251](https://github.com/databricks/dbt-databricks/issues/1251))
 
 ### Under the Hood

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

@@ -181,9 +181,18 @@ replace on ({{ replace_on_expr }})
 
   {#-- Build WHERE clause for DELETE statement --#}
   {%- set delete_conditions = [] -%}
-  {%- for key in unique_keys -%}
+  {%- if unique_keys | length > 1 -%}
+    {#-- a row-valued IN raises DELTA_UNSUPPORTED_MULTI_COL_IN_PREDICATE; correlate on
+         the whole tuple instead so unmatched key combinations are not deleted (issue #1611) --#}
+    {%- set correlation_conditions = [] -%}
+    {%- for key in unique_keys -%}
+      {%- do correlation_conditions.append(target_relation ~ '.' ~ adapter.quote(key) ~ ' <=> ' ~ source_relation ~ '.' ~ adapter.quote(key)) -%}
+    {%- endfor -%}
+    {%- do delete_conditions.append('EXISTS (SELECT 1 FROM ' ~ source_relation ~ ' WHERE ' ~ correlation_conditions | join(' AND ') ~ ')') -%}
+  {%- else -%}
+    {%- set key = unique_keys[0] -%}
     {%- do delete_conditions.append(target_relation ~ '.' ~ adapter.quote(key) ~ ' IN (SELECT ' ~ adapter.quote(key) ~ ' FROM ' ~ source_relation ~ ')') -%}
-  {%- endfor -%}
+  {%- 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,replaced
+1,red,updated
+2,blue,updated
+2,red,goodbye
+"""
+
 delete_insert_update_schema_expected = """id
 1
 2
@@ -442,6 +449,48 @@
 {% 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 %}
+
+-- (1, blue) is an exact key match and must be replaced with its new payload.
+-- (1, red) and (2, blue) only bait a per-column match; matching each column on
+-- its own would wrongly delete both existing rows, so they must only insert.
+-- (2, red) is absent from this run and so must survive untouched.
+select cast(1 as bigint) as id, 'blue' as color, 'replaced' as msg
+union all
+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,41 @@ 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 correlates 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 (
+            "exists (select 1 from source where target.`a` <=> source.`a`"
+            " and target.`b` <=> source.`b`)"
+        ) in clean_delete
+        assert clean_delete.startswith("delete from target where exists")
 
-    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 EXISTS clause, 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 (
+            "exists (select 1 from source where target.`a` <=> source.`a`"
+            " and target.`b` <=> source.`b`)"
+        ) in clean_delete
+        assert clean_delete.endswith("and a > 1")
 
-        # 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"""