fix: deterministically name unnamed PK/FK so incremental constraints don't churn or collide

CHANGELOG.md

@@ -9,6 +9,7 @@
 - Fix managed Iceberg Python models failing with `MANAGED_TABLE_FORMAT` by emitting `.format("iceberg")` instead of the `parquet` sentinel from `resolve_file_format` (thanks @Divya-Kovvuru-0802!) ([#1593](https://github.com/databricks/dbt-databricks/pull/1593) resolves [#1591](https://github.com/databricks/dbt-databricks/issues/1591))
 - Quote generated column identifiers in incremental strategies so non-ASCII column names no longer fail on subsequent runs ([#1595](https://github.com/databricks/dbt-databricks/pull/1595) resolves [#1594](https://github.com/databricks/dbt-databricks/issues/1594))
 - Handle missing or empty view-definition metadata when creating materialized views from streaming tables or newly-created materialized views (thanks @aarushisingh04!) ([#1462](https://github.com/databricks/dbt-databricks/pull/1462) resolves [#1459](https://github.com/databricks/dbt-databricks/issues/1459))
+- Fix unnamed primary and foreign keys churning on every incremental run, and two or more unnamed foreign keys to the same parent failing with `DELTA_CONSTRAINT_ALREADY_EXISTS`. dbt now gives an unnamed PK/FK a deterministic name (its full identity, including a foreign key's referenced columns) generated identically on the create and incremental paths — including V2 inline `CREATE TABLE` and bare-parent FK configs — so the model and catalog agree and the diff is a no-op. Existing unnamed foreign keys are renamed once on the next incremental run (a no-op drop/re-add, no cascade). An unnamed key's `expression`/target edit is reconciled via that name; a named key's `expression` edit still needs `--full-refresh` (see #1552). ([#1561](https://github.com/databricks/dbt-databricks/pull/1561) resolves [#1333](https://github.com/databricks/dbt-databricks/issues/1333) and [#1344](https://github.com/databricks/dbt-databricks/issues/1344))
 
 ### Under the Hood
 

dbt/adapters/databricks/constraints.py

@@ -1,3 +1,4 @@
+import hashlib
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from typing import Any, ClassVar, Optional, TypeVar
@@ -162,6 +163,30 @@ def is_enforced(constraint: ColumnLevelConstraint) -> bool:
     ]
 
 
+def _local_md5(value: str) -> str:
+    # Same digest as Jinja local_md5.
+    return hashlib.md5(value.encode("utf-8")).hexdigest()
+
+
+def synthesize_constraint_name(constraint: TypedConstraint, relation_identifier: str) -> str:
+    """Deterministic name for an unnamed PK/FK; mirrors relations/constraints.sql."""
+    if isinstance(constraint, PrimaryKeyConstraint):
+        hash_input = f"primary_key;{relation_identifier};{constraint.columns};"
+        if constraint.expression:
+            hash_input += f"{constraint.expression};"
+        return _local_md5(hash_input)
+    if isinstance(constraint, ForeignKeyConstraint):
+        if constraint.expression:
+            return _local_md5(f"foreign_key;{relation_identifier};{constraint.expression};")
+        hash_input = f"foreign_key;{relation_identifier};{constraint.columns};{constraint.to};"
+        if constraint.to_columns:
+            hash_input += f"{constraint.to_columns};"
+        return _local_md5(hash_input)
+    raise DbtValidationError(
+        f"Cannot synthesize a name for constraint type: {type(constraint).__name__}"
+    )
+
+
 def process_constraint(constraint: TypedConstraint) -> Optional[str]:
     if validate_constraint(constraint):
         return constraint.render()

dbt/adapters/databricks/relation.py

@@ -13,7 +13,13 @@
 from dbt_common.exceptions import DbtRuntimeError
 from dbt_common.utils import filter_null_values
 
-from dbt.adapters.databricks.constraints import TypedConstraint, process_constraint
+from dbt.adapters.databricks.constraints import (
+    ForeignKeyConstraint,
+    PrimaryKeyConstraint,
+    TypedConstraint,
+    process_constraint,
+    synthesize_constraint_name,
+)
 from dbt.adapters.databricks.logging import logger
 from dbt.adapters.databricks.utils import remove_undefined
 
@@ -262,6 +268,11 @@ def enrich(self, constraints: list[TypedConstraint]) -> "DatabricksRelation":
         return copy
 
     def render_constraints_for_create(self) -> str:
+        for constraint in self.create_constraints:
+            if constraint.name is None and isinstance(
+                constraint, (PrimaryKeyConstraint, ForeignKeyConstraint)
+            ):
+                constraint.name = synthesize_constraint_name(constraint, self.identifier)
         processed = map(process_constraint, self.create_constraints)
         return ", ".join(c for c in processed if c is not None)
 

dbt/adapters/databricks/relation_configs/constraints.py

@@ -13,6 +13,7 @@
     PrimaryKeyConstraint,
     TypedConstraint,
     parse_constraints,
+    synthesize_constraint_name,
 )
 from dbt.adapters.databricks.relation_configs.base import (
     DatabricksComponentConfig,
@@ -251,6 +252,12 @@ def from_relation_config(cls, relation_config: RelationConfig) -> ConstraintsCon
 
         non_nulls, other_constraints = parse_constraints(columns, constraints)
 
+        for constraint in other_constraints:
+            if constraint.name is None and isinstance(
+                constraint, (PrimaryKeyConstraint, ForeignKeyConstraint)
+            ):
+                constraint.name = synthesize_constraint_name(constraint, relation_config.identifier)
+
         return ConstraintsConfig(
             set_non_nulls=set(non_nulls),
             set_constraints=set(other_constraints),

dbt/include/databricks/macros/relations/constraints.sql

@@ -225,6 +225,8 @@
       {% set joined_names = quoted_names|join(", ") %}
 
       {% set parent = constraint.get('to') %}
+      {# Hash raw `to` so bare parents match synthesize_constraint_name. #}
+      {% set raw_parent = parent %}
       {% if not parent %}
         {{ exceptions.raise_compiler_error('No parent table defined for foreign key: ' ~ expression) }}
       {% endif %}
@@ -233,17 +235,21 @@
         {% set parent = parent_relation.render() %}
       {% endif %}
 
+      {% set parent_columns = constraint.get('to_columns') %}
       {% if not name %}
         {% if local_md5 %}
           {{ exceptions.warn("Constraint of type " ~ type ~ " with no `name` provided. Generating hash instead for relation " ~ relation.identifier) }}
-          {%- set name = local_md5("foreign_key;" ~ relation.identifier ~ ";" ~ column_names ~ ";" ~ parent ~ ";") -%}
+          {%- set hash_input = "foreign_key;" ~ relation.identifier ~ ";" ~ column_names ~ ";" ~ raw_parent ~ ";" -%}
+          {%- if parent_columns -%}
+            {%- set hash_input = hash_input ~ parent_columns ~ ";" -%}
+          {%- endif -%}
+          {%- set name = local_md5(hash_input) -%}
         {% else %}
           {{ exceptions.raise_compiler_error("Constraint of type " ~ type ~ " with no `name` provided, and no md5 utility.") }}
-        {% endif %}    
+        {% endif %}
       {% endif %}
 
       {% set stmt = "alter table " ~ relation.render() ~ " add constraint " ~ name ~ " foreign key(" ~ joined_names ~ ") references " ~ parent %}
-      {% set parent_columns = constraint.get('to_columns') %}
       {% if parent_columns %}
         {% set quoted_parent_columns = [] %}
         {% for parent_column in parent_columns %}

tests/functional/adapter/constraints/fixtures.py

@@ -379,7 +379,6 @@
         constraints:
           - type: not_null
           - type: primary_key
-            name: pk_rely_parent
             expression: RELY
   - name: rely_child
     config:
@@ -411,6 +410,55 @@
 select 1 as parent_n, 10 as child_id
 """
 
+incremental_multiple_fk_schema_yml = """
+version: 2
+models:
+  - name: multi_fk_parent
+    config:
+      materialized: table
+      contract:
+        enforced: true
+    columns:
+      - name: id
+        data_type: int
+        constraints:
+          - type: not_null
+          - type: primary_key
+            name: pk_multi_fk_parent
+  - name: multi_fk_child
+    config:
+      materialized: incremental
+      unique_key: child_id
+      on_schema_change: append_new_columns
+      contract:
+        enforced: true
+    columns:
+      - name: child_id
+        data_type: int
+      - name: parent_a
+        data_type: int
+        constraints:
+          - type: foreign_key
+            to: ref('multi_fk_parent')
+            to_columns: ["id"]
+      - name: parent_b
+        data_type: int
+        constraints:
+          - type: foreign_key
+            to: ref('multi_fk_parent')
+            to_columns: ["id"]
+"""
+
+incremental_multiple_fk_parent_sql = """
+select 1 as id
+"""
+
+incremental_multiple_fk_child_sql = """
+-- depends_on: {{ ref('multi_fk_parent') }}
+
+select 1 as child_id, 1 as parent_a, 1 as parent_b
+"""
+
 
 def _incremental_contract_off_pk_schema_yml(enforced):
     return f"""
@@ -463,3 +511,49 @@ def _incremental_contract_off_pk_schema_yml(enforced):
       - name: color
         data_type: string
 """
+
+incremental_v2_unnamed_pk_cascade_schema_yml = """
+version: 2
+models:
+  - name: v2_unnamed_pk_parent
+    config:
+      materialized: incremental
+      unique_key: n
+      on_schema_change: append_new_columns
+      contract:
+        enforced: true
+    columns:
+      - name: n
+        data_type: int
+        constraints:
+          - type: not_null
+          - type: primary_key
+  - name: v2_unnamed_pk_child
+    config:
+      materialized: table
+      contract:
+        enforced: true
+    constraints:
+      - type: foreign_key
+        name: fk_v2_unnamed_pk_child
+        columns: ["parent_n"]
+        to: ref('v2_unnamed_pk_parent')
+        to_columns: ["n"]
+    columns:
+      - name: parent_n
+        data_type: int
+        constraints:
+          - type: not_null
+      - name: child_id
+        data_type: int
+"""
+
+incremental_v2_unnamed_pk_parent_sql = """
+select 1 as n
+"""
+
+incremental_v2_unnamed_pk_child_sql = """
+-- depends_on: {{ ref('v2_unnamed_pk_parent') }}
+
+select 1 as parent_n, 10 as child_id
+"""

tests/functional/adapter/constraints/test_constraints.py

@@ -222,10 +222,12 @@ def test_foreign_key_constraint(self, project):
 
 
 @pytest.mark.skip_profile("databricks_cluster")
-class TestIncrementalRelyConstraintReconciliation:
-    """A RELY expression on a primary key cannot be read back from information_schema, so it
-    must not trigger constraint reconciliation on an incremental run. Otherwise the parent PK
-    is dropped with CASCADE every run, silently dropping the child's foreign key (#1513).
+class TestIncrementalPrimaryKeyConstraintReconciliation:
+    """A primary key's non-round-trippable fields must not trigger reconciliation on an incremental
+    run. RELY (#1513) is not readable from information_schema, and an unnamed PK (#1333) is given a
+    server-assigned name the model lacks; treating either as a change drops the PK with CASCADE
+    every run, silently dropping the child's foreign key. The parent PK here is both unnamed and
+    RELY, so this guards both fixes.
     """
 
     @pytest.fixture(scope="class")
@@ -251,16 +253,62 @@ def _foreign_key_names(self, project):
         )
         return {row[0] for row in rows}
 
-    def test_rely_pk_reconcile_keeps_dependent_foreign_key(self, project):
+    def test_unnamed_rely_pk_reconcile_keeps_dependent_foreign_key(self, project):
         util.run_dbt(["build"])
         assert "fk_rely_child" in self._foreign_key_names(project)
 
-        # A plain incremental re-run of the parent must not reconcile its RELY PK.
+        # A plain incremental re-run of the parent must not reconcile its unnamed RELY PK.
         util.run_dbt(["run", "--select", "rely_parent"])
 
         assert "fk_rely_child" in self._foreign_key_names(project)
 
 
+@pytest.mark.skip_profile("databricks_cluster")
+class TestIncrementalMultipleUnnamedForeignKeys:
+    """Two unnamed foreign keys to the same parent must both survive an incremental run. Without a
+    deterministic name, the model's name=None never matches the catalog's server-assigned name, so
+    the diff drops and re-adds both every run, and the unnamed re-adds collide on the server-derived
+    name with DELTA_CONSTRAINT_ALREADY_EXISTS (#1344).
+    """
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {"flags": {"use_materialization_v2": False}}
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "schema.yml": override_fixtures.incremental_multiple_fk_schema_yml,
+            "multi_fk_parent.sql": override_fixtures.incremental_multiple_fk_parent_sql,
+            "multi_fk_child.sql": override_fixtures.incremental_multiple_fk_child_sql,
+        }
+
+    def _foreign_key_columns(self, project):
+        rows = project.run_sql(
+            """
+            SELECT kcu.column_name
+            FROM {database}.information_schema.key_column_usage kcu
+            JOIN {database}.information_schema.table_constraints tc
+              ON kcu.constraint_name = tc.constraint_name
+             AND kcu.constraint_schema = tc.constraint_schema
+            WHERE tc.constraint_schema = '{schema}'
+              AND tc.table_name = 'multi_fk_child'
+              AND tc.constraint_type = 'FOREIGN KEY'
+            """,
+            fetch="all",
+        )
+        return {row[0] for row in rows}
+
+    def test_multiple_unnamed_fks_survive_incremental_run(self, project):
+        util.run_dbt(["build"])
+        assert {"parent_a", "parent_b"} <= self._foreign_key_columns(project)
+
+        # The incremental re-run must not drop and re-add the unnamed FKs (they would collide).
+        util.run_dbt(["run", "--select", "multi_fk_child"])
+
+        assert {"parent_a", "parent_b"} <= self._foreign_key_columns(project)
+
+
 @pytest.mark.skip_profile("databricks_cluster")
 class TestIncrementalContractOffPreservesConstraints(RerunSafeMixin):
     """Constraints are reconciled only when the contract is enforced; when unenforced, a no-op."""
@@ -306,3 +354,39 @@ def test_contract_off_incremental_preserves_existing_pk(self, project):
 
         # The existing PK must be left untouched, not silently dropped.
         assert "pk_contract_off" in self._primary_key_names(project)
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestV2IncrementalUnnamedPrimaryKeyReconciliation:
+    """V2: unnamed parent PK re-run must leave the dependent FK intact."""
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {"flags": {"use_materialization_v2": True}}
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "schema.yml": override_fixtures.incremental_v2_unnamed_pk_cascade_schema_yml,
+            "v2_unnamed_pk_parent.sql": override_fixtures.incremental_v2_unnamed_pk_parent_sql,
+            "v2_unnamed_pk_child.sql": override_fixtures.incremental_v2_unnamed_pk_child_sql,
+        }
+
+    def _foreign_key_names(self, project):
+        rows = project.run_sql(
+            """
+            SELECT constraint_name
+            FROM {database}.information_schema.referential_constraints
+            WHERE constraint_schema = '{schema}'
+            """,
+            fetch="all",
+        )
+        return {row[0] for row in rows}
+
+    def test_unnamed_pk_reconcile_keeps_dependent_foreign_key(self, project):
+        util.run_dbt(["build"])
+        assert "fk_v2_unnamed_pk_child" in self._foreign_key_names(project)
+
+        util.run_dbt(["run", "--select", "v2_unnamed_pk_parent"])
+
+        assert "fk_v2_unnamed_pk_child" in self._foreign_key_names(project)