Fix: Retain multiple foreign-keys from one table to another after an incremental run (#1296)

<!-- Please review our pull request review process in CONTRIBUTING.md
before your proceed. -->



<!---
Include the number of the issue addressed by this PR above if
applicable.

  Example:
    resolves #1234

Please review our pull request review process in CONTRIBUTING.md before
your proceed.
-->

### Description

Encountered a bug when:
  * `"use_materialization_v2": False`
  * `config.contract.enforced: true`
* Multiple foreign keys are specified from one table to another (e.g. a
fact table with multiple date values, each of which has a foreign key
constrain to a common data dimension)
* An initial full run of the model with the FK constraints is carried
out, followed by an incremental run.

The bug is that on the initial full run the multiple FK constraints to
the common table are created correctly on the Databricks table, but then
when an incremental run is carried out these FKs are removed from the
Databricks table. Single FKs to other tables seem to be retained (this
isn't replicated in the test case in the interest of keeping it
minimal).

The bug is replicated in the integration test
[TestIncrementalMultipleFKsToSameTableNoV2](https://github.com/databricks/dbt-databricks/compare/main...Ewan-Keith:dbt-databricks:consistenty-persist-databricks-constraints?expand=1#diff-64ad276353f7ed237d5072874d1fe18c67310f90493ce7dc8a9fc69756925335R380),
before the fix is applied this passes the first assertion (after the
initial full run) but fails the second assertion (after the incremental
run) with an empty set of constraints returned. A matching test is also
present replicating the logic when using materialization_v2
[TestIncrementalMultipleFKsToSameTable](https://github.com/databricks/dbt-databricks/compare/main...Ewan-Keith:dbt-databricks:consistenty-persist-databricks-constraints?expand=1#diff-64ad276353f7ed237d5072874d1fe18c67310f90493ce7dc8a9fc69756925335R334)
which confirms the issue is not seen with v2 materialization (this
passes without any actual code changes).

The fix I've landed on in
`dbt/include/databricks/macros/materializations/incremental/incremental.sql`
is to treat constraints as one of the configuration changes that is
explicitly applied on an incremental run when changes are detected (the
same as for `tags`, `tblproperties`, and `liquid_clustering`). This gets
the failing test passing.

I don't think this is necessarily a root-cause fix, I think the root
issue is likely a change in constraints being falsely detected in this
instance, but I wasn't sure where to start on that, and this fix seems
sensible (even if the root issue is resolved it seems likely we'd want
changes to constraints to be applied without forcing a full-refresh
where this is possible?). There's a unit test checking if there are any
obvious bugs in the constraint diff checker but this doesn't turn
anything up (passes as expected).


### 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: Ewan Keith <ewan.keith@kaluza.com>
Co-authored-by: Ben Cassell <98852248+benc-db@users.noreply.github.com>

Author: Ewan-Keith

CHANGELOG.md

@@ -1,3 +1,9 @@
+## dbt-databricks 1.11.5 (TBD)
+
+### Fixes
+
+- Fix foreign-key on an incremental table to a primary key on a non-incremental table being lost after incremental run
+
 ## dbt-databricks 1.11.4 (Jan 12, 2026)
 
 ### Features

dbt/adapters/databricks/constraints.py

@@ -119,12 +119,14 @@ def _validate(self) -> None:
             )
 
     def _render_suffix(self) -> str:
-        suffix = f"FOREIGN KEY ({', '.join(self.columns)})"
         if self.expression:
-            suffix += f" {self.expression}"
-        else:
-            suffix += f" REFERENCES {self.to} ({', '.join(self.to_columns)})"
-        return suffix
+            if self.expression.strip().startswith("("):
+                return f"FOREIGN KEY {self.expression}"
+            return f"FOREIGN KEY ({', '.join(self.columns)}) {self.expression}"
+        return (
+            f"FOREIGN KEY ({', '.join(self.columns)}) REFERENCES "
+            + f"{self.to} ({', '.join(self.to_columns)})"
+        )
 
 
 class CheckConstraint(TypedConstraint):

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

@@ -177,6 +177,7 @@
         {% set tags = _configuration_changes.changes.get("tags", None) %}
         {% set tblproperties = _configuration_changes.changes.get("tblproperties", None) %}
         {% set liquid_clustering = _configuration_changes.changes.get("liquid_clustering") %}
+        {% set constraints = _configuration_changes.changes.get("constraints") %}
         {% if tags is not none %}
           {% do apply_tags(target_relation, tags.set_tags) %}
         {%- endif -%}
@@ -186,6 +187,10 @@
         {% if liquid_clustering is not none %}
           {% do apply_liquid_clustered_cols(target_relation, liquid_clustering) %}
         {% endif %}
+        {#- Incremental constraint application requires information_schema access (see fetch_*_constraints macros) -#}
+        {% if constraints and not target_relation.is_hive_metastore() %}
+          {{ apply_constraints(target_relation, constraints) }}
+        {% endif %}
       {%- endif -%}
       {% do persist_docs(target_relation, model, for_relation=True) %}
     {%- endif -%}

tests/functional/adapter/incremental/fixtures.py

@@ -1109,3 +1109,55 @@ def model(dbt, spark):
         to_columns: [id]
         warn_unenforced: false
 """
+
+non_incremental_target_of_fk = """
+{{ config(
+    materialized='table',
+) }}
+
+SELECT 'a' AS str_key;
+"""
+
+incremental_fk_sql = """
+{{ config(
+    materialized='incremental',
+    unique_key=['fk_col'],
+    incremental_strategy='delete+insert',
+    on_schema_change='fail'
+) }}
+
+SELECT
+    'a' AS fk_col
+"""
+
+incremental_fk_on_non_incremental_target_schema_yml = """
+version: 2
+
+models:
+  - name: non_incremental_target_of_fk
+    config:
+      contract:
+        enforced: true
+    columns:
+      - name: str_key
+        data_type: string
+        constraints:
+          - type: not_null
+          - type: primary_key
+            name: pk_target_key
+            warn_unenforced: false
+
+  - name: incremental_fk
+    config:
+      contract:
+        enforced: true
+    columns:
+      - name: fk_col
+        data_type: string
+        constraints:
+          - type: foreign_key
+            to: ref('non_incremental_target_of_fk')
+            to_columns: ["str_key"]
+            name: fk
+            warn_unenforced: false
+"""

tests/functional/adapter/incremental/test_incremental_constraints.py

@@ -328,3 +328,79 @@ def test_remove_foreign_key_constraint(self, project):
         util.run_dbt(["run"])
         referential_constraints = project.run_sql(referential_constraint_sql, fetch="all")
         assert len(referential_constraints) == 0
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestIncrementalFkTargetNonIncrementalIsRetained:
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {"use_materialization_v2": True},
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "non_incremental_target_of_fk.sql": fixtures.non_incremental_target_of_fk,
+            "incremental_fk.sql": fixtures.incremental_fk_sql,
+            "schema.yml": fixtures.incremental_fk_on_non_incremental_target_schema_yml,
+        }
+
+    def test_multiple_fks_to_same_table_persist_after_incremental(self, project):
+        expected_constraints = {
+            ("fk", "pk_target_key"),
+        }
+
+        # Initial run - create tables with constraints
+        util.run_dbt(["run"])
+
+        referential_constraints = project.run_sql(referential_constraint_sql, fetch="all")
+
+        constraints = {(row[0], row[1]) for row in referential_constraints}
+        assert constraints == expected_constraints
+
+        # Incremental run - this should NOT lose any foreign keys
+        util.run_dbt(["run"])
+
+        referential_constraints_after = project.run_sql(referential_constraint_sql, fetch="all")
+
+        constraint_names_after = {(row[0], row[1]) for row in referential_constraints_after}
+        assert constraint_names_after == expected_constraints
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestIncrementalFkTargetNonIncrementalIsRetainedNoV2:
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {"use_materialization_v2": False},
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "non_incremental_target_of_fk.sql": fixtures.non_incremental_target_of_fk,
+            "incremental_fk.sql": fixtures.incremental_fk_sql,
+            "schema.yml": fixtures.incremental_fk_on_non_incremental_target_schema_yml,
+        }
+
+    def test_multiple_fks_to_same_table_persist_after_incremental(self, project):
+        expected_constraints = {
+            ("fk", "pk_target_key"),
+        }
+
+        # Initial run - create tables with constraints
+        util.run_dbt(["run"])
+
+        referential_constraints = project.run_sql(referential_constraint_sql, fetch="all")
+
+        constraints = {(row[0], row[1]) for row in referential_constraints}
+        assert constraints == expected_constraints
+
+        # Incremental run - this should NOT lose any foreign keys
+        util.run_dbt(["run"])
+
+        referential_constraints_after = project.run_sql(referential_constraint_sql, fetch="all")
+
+        constraint_names_after = {(row[0], row[1]) for row in referential_constraints_after}
+        assert constraint_names_after == expected_constraints