diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1e27d06b2..d8d2664f7 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/dbt/include/databricks/macros/materializations/incremental/strategies.sql b/dbt/include/databricks/macros/materializations/incremental/strategies.sql
index 7b2c73c8d..44eba7fee 100644
--- a/dbt/include/databricks/macros/materializations/incremental/strategies.sql
+++ b/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 -%}
diff --git a/tests/functional/adapter/incremental/fixtures.py b/tests/functional/adapter/incremental/fixtures.py
index ce559bb60..81e1c793c 100644
--- a/tests/functional/adapter/incremental/fixtures.py
+++ b/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',
diff --git a/tests/functional/adapter/incremental/test_incremental_strategies.py b/tests/functional/adapter/incremental/test_incremental_strategies.py
index 238bd3166..1104cd0ed 100644
--- a/tests/functional/adapter/incremental/test_incremental_strategies.py
+++ b/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):
diff --git a/tests/unit/macros/materializations/incremental/test_delete_insert.py b/tests/unit/macros/materializations/incremental/test_delete_insert.py
index c2e09ebc1..81957bda1 100644
--- a/tests/unit/macros/materializations/incremental/test_delete_insert.py
+++ b/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"""