Fix: hard_deletes invalidate incorrectly invalidating active records in snapshots (#1283)

Resolves #1281

### Description

When running a snapshot with `hard_deletes: invalidate`, records that
still exist in the source table were being incorrectly marked as
invalidated.

**Root Cause:** The `when not matched by source` clause (introduced in
1.11.1) matches unchanged records because the staging table only
contains *changed* records, not all records.

**Solution:** Removed the `when not matched by source` clause. The
existing `deletes` CTE already properly handles deletions via
`dbt_change_type = 'delete'` and the correct `dbt_scd_id`.
We reproduced and verified the fix using the exact example provided in
the issue.

**Tested on:** SQL Warehouse with Unity Catalog

### 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: Stefan Zurborg <stefan.zurborg@fielmann.com>
Co-authored-by: Ben Cassell <98852248+benc-db@users.noreply.github.com>

Author: Zurbste

CHANGELOG.md

@@ -1,7 +1,10 @@
 ## dbt-databricks 1.11.4 (TBD)
 
 ### Features
-- Add `query_id` to `SQLQueryStatus` events to improve query tracing and debugging 
+- Add `query_id` to `SQLQueryStatus` events to improve query tracing and debugging
+
+### Fixes
+- Fix `hard_deletes: invalidate` incorrectly invalidating active records in snapshots (thanks @Zurbste!) ([#1281](https://github.com/databricks/dbt-databricks/issues/1281)) 
 
 ## dbt-databricks 1.11.3 (Dec 5, 2025)
 

dbt/include/databricks/macros/materializations/snapshot_helpers.sql

@@ -30,13 +30,6 @@
         and DBT_INTERNAL_SOURCE.{{ adapter.quote('dbt_change_type') }} = 'insert'
         then insert ({{ insert_cols_csv }})
         values ({{ insert_cols_csv }})
-
-    {%- if invalidate_hard_deletes %}
-    when not matched by source
-        and DBT_INTERNAL_DEST.{{ columns.dbt_valid_to }} is null
-        then update set
-            {{ columns.dbt_valid_to }} = current_timestamp()
-    {%- endif %}
     ;
 {% endmacro %}
 

tests/functional/adapter/simple_snapshot/test_hard_deletes.py

@@ -164,21 +164,32 @@ def test_hard_delete_invalidate(self, project):
         )
 
         # Verify deleted records (3, 4) have dbt_valid_to set (not NULL)
+        # and non-deleted records (1, 2, 5) have dbt_valid_to = NULL
         # Snapshot columns: id, name, city, updated_at, dbt_scd_id,
         # dbt_updated_at, dbt_valid_from, dbt_valid_to
         # dbt_valid_to is the last column (index -1)
         invalidated_count = 0
+        active_count = 0
         for record in final_records:
-            if record[0] in [3, 4]:  # id column
-                # dbt_valid_to should NOT be NULL
-                dbt_valid_to = record[-1]  # last column
+            record_id = record[0]  # id column
+            dbt_valid_to = record[-1]  # last column
+            if record_id in [3, 4]:
+                # Deleted records: dbt_valid_to should NOT be NULL
                 assert dbt_valid_to is not None, (
-                    f"Record id={record[0]} should have dbt_valid_to set "
+                    f"Record id={record_id} should have dbt_valid_to set "
                     f"with hard_deletes='invalidate', but got {dbt_valid_to}"
                 )
                 invalidated_count += 1
+            elif record_id in [1, 2, 5]:
+                # Non-deleted records: dbt_valid_to should be NULL (still active)
+                assert dbt_valid_to is None, (
+                    f"Record id={record_id} should have NULL dbt_valid_to "
+                    f"(still active), but got {dbt_valid_to}"
+                )
+                active_count += 1
 
         assert invalidated_count == 2, f"Expected 2 invalidated records, found {invalidated_count}"
+        assert active_count == 3, f"Expected 3 active records, found {active_count}"
 
 
 class TestHardDeleteNewRecord(BaseHardDeleteTest):