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

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
@@ -8,6 +14,7 @@
 
 - Fix `hard_deletes: invalidate` incorrectly invalidating active records in snapshots (thanks @Zurbste!) ([#1281](https://github.com/databricks/dbt-databricks/issues/1281))
 - Fix serverless Python model environment configuration: use `environment_version` instead of deprecated `client` field. Users can now specify custom environment versions via `python_job_config.environments`. ([#1286](https://github.com/databricks/dbt-databricks/pull/1286))
+- Fix foreign-key on an incremental table to a primary key on a non-incremental table being lost after incremental run
 
 ### Under the Hood
 

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