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

CHANGELOG.md

@@ -9,7 +9,8 @@
 - Honor the `expression` field on `primary_key` constraints on the V1 materialization path. A primary key declared with `expression: RELY` (or any trailing clause) previously had its expression silently dropped. ([#1551](https://github.com/databricks/dbt-databricks/pull/1551))
 - Apply column-level `databricks_tags` for incremental models on the V1 materialization path ([#1520](https://github.com/databricks/dbt-databricks/pull/1520) closes [#1307](https://github.com/databricks/dbt-databricks/issues/1307))
 - Raise a `DbtRuntimeError` when a Python model job run terminates with a non-success `result_state` (e.g. `FAILED`/`TIMEDOUT`) instead of returning silently ([#1477](https://github.com/databricks/dbt-databricks/pull/1477))
-- Fix PK/FK constraints declaring an `expression` (e.g. `RELY`) being dropped and re-added on every incremental run. **Regression:** changing the `expression` on an existing PK/FK (`RELY`↔`NORELY`, or an expression-form FK's target) is no longer applied on incremental runs — use `--full-refresh`. ([#1552](https://github.com/databricks/dbt-databricks/pull/1552) closes [#1513](https://github.com/databricks/dbt-databricks/issues/1513))
+- Fix PK/FK constraints declaring an `expression` (e.g. `RELY`) being dropped and re-added on every incremental run. **Regression:** changing the `expression` on an existing **named** PK/FK (`RELY`↔`NORELY`, or an expression-form FK's target) is no longer applied on incremental runs — use `--full-refresh`. (Unnamed keys are unaffected: their generated name encodes the `expression`, so such a change is detected and reconciled.) ([#1552](https://github.com/databricks/dbt-databricks/pull/1552) closes [#1513](https://github.com/databricks/dbt-databricks/issues/1513))
+- 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, 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). ([#1561](https://github.com/databricks/dbt-databricks/pull/1561) closes [#1333](https://github.com/databricks/dbt-databricks/issues/1333) and [#1344](https://github.com/databricks/dbt-databricks/issues/1344))
 - Honor `incremental_apply_config_changes` in the V1 incremental merge path, allowing users to skip metadata diff queries (tags, column_tags, constraints, column_masks, tblproperties, describe_extended) when set to `false`. Matches the existing V2 behavior. ([1467](https://github.com/databricks/dbt-databricks/pull/1467) partially solves [#1402](https://github.com/databricks/dbt-databricks/issues/1402))
 - Fix column-level `databricks_tags` on Unity Catalog views updated via `ALTER` (`view_update_via_alter: true`) ([#1526](https://github.com/databricks/dbt-databricks/pull/1526) closes [#1525](https://github.com/databricks/dbt-databricks/issues/1525))
 - Apply `tblproperties` to `metric_view` models at create time, not only on a later alter/replace run ([#1530](https://github.com/databricks/dbt-databricks/pull/1530) closes [#1527](https://github.com/databricks/dbt-databricks/issues/1527))

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,32 @@ def is_enforced(constraint: ColumnLevelConstraint) -> bool:
     ]
 
 
+def _local_md5(value: str) -> str:
+    # Matches dbt's `local_md5` Jinja helper (and the create-time constraint macro), so a name
+    # computed here is byte-identical to the one already written to the catalog.
+    return hashlib.md5(value.encode("utf-8")).hexdigest()
+
+
+def synthesize_constraint_name(constraint: TypedConstraint, relation_identifier: str) -> str:
+    """Name an unnamed PK/FK; mirrors the create-time macro (``relations/constraints.sql``) so the
+    model side matches the name already in the catalog."""
+    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_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

@@ -233,17 +233,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 ~ ";" ~ 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:
@@ -410,3 +409,52 @@
 
 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
+"""

tests/functional/adapter/constraints/test_constraints.py

@@ -221,10 +221,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")
@@ -250,11 +252,57 @@ 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)

tests/unit/macros/relations/test_constraint_macros.py

@@ -1,5 +1,14 @@
-import pytest
+import re
 
+import pytest
+from dbt_common.contracts.constraints import ConstraintType
+
+from dbt.adapters.databricks.constraints import (
+    ForeignKeyConstraint,
+    PrimaryKeyConstraint,
+    _local_md5,
+    synthesize_constraint_name,
+)
 from tests.unit.macros.base import MacroTestBase
 
 
@@ -474,3 +483,119 @@ def test_macros_get_constraint_sql_custom_missing_expression(self, template_bund
         }
         r = self.render_constraint_sql(template_bundle, constraint, model)
         assert "raise_compiler_error" in r
+
+
+class TestConstraintNameParity(MacroTestBase):
+    """The create-time macro (``get_constraint_sql``) and the Python ``synthesize_constraint_name``
+    are two implementations of one naming scheme: the macro names an unnamed PK/FK at create time,
+    and the Python function reproduces that name on the model side so the incremental diff matches
+    the catalog instead of churning + colliding (#1333, #1344). Until the macro is retired and the
+    two are unified into one code path, these tests pin them together so neither can drift.
+
+    The macro test harness mocks ``local_md5``; here we inject the real wrapper on the macro side so
+    both implementations hash with the same function -- the test then verifies the hash *inputs*
+    (the real drift risk) agree.
+    """
+
+    @pytest.fixture
+    def template_name(self) -> str:
+        return "constraints.sql"
+
+    @pytest.fixture
+    def macro_folders_to_load(self) -> list:
+        return ["macros/relations", "macros"]
+
+    @pytest.fixture
+    def model(self):
+        columns = {name: {"name": name, "data_type": "int"} for name in ("a", "b", "c")}
+        return {"columns": columns}
+
+    def _macro_name(self, template_bundle, constraint, *args):
+        # Swap the harness's mock local_md5 for the real one so the macro yields the real name.
+        template_bundle.context["local_md5"] = _local_md5
+        rendered = self.run_macro_raw(
+            template_bundle.template,
+            "get_constraint_sql",
+            template_bundle.relation,
+            constraint,
+            *args,
+        )
+        match = re.search(r"add constraint (\S+) ", rendered)
+        assert match, f"no constraint name found in rendered macro output: {rendered}"
+        return match.group(1)
+
+    def test_parity__foreign_key_single_column(self, template_bundle, model):
+        constraint = {
+            "type": "foreign_key",
+            "columns": ["a"],
+            "to": "`c`.`s`.`parent`",
+            "to_columns": ["id"],
+        }
+        py_name = synthesize_constraint_name(
+            ForeignKeyConstraint(
+                type=ConstraintType.foreign_key,
+                columns=["a"],
+                to="`c`.`s`.`parent`",
+                to_columns=["id"],
+            ),
+            template_bundle.relation.identifier,
+        )
+        assert self._macro_name(template_bundle, constraint, model) == py_name
+
+    def test_parity__foreign_key_multiple_columns(self, template_bundle, model):
+        constraint = {
+            "type": "foreign_key",
+            "columns": ["a", "b"],
+            "to": "`c`.`s`.`parent`",
+            "to_columns": ["x", "y"],
+        }
+        py_name = synthesize_constraint_name(
+            ForeignKeyConstraint(
+                type=ConstraintType.foreign_key,
+                columns=["a", "b"],
+                to="`c`.`s`.`parent`",
+                to_columns=["x", "y"],
+            ),
+            template_bundle.relation.identifier,
+        )
+        assert self._macro_name(template_bundle, constraint, model) == py_name
+
+    def test_parity__foreign_key_expression_form(self, template_bundle, model):
+        constraint = {
+            "type": "foreign_key",
+            "columns": ["a"],
+            "expression": "(a) REFERENCES `c`.`s`.`parent`",
+        }
+        py_name = synthesize_constraint_name(
+            ForeignKeyConstraint(
+                type=ConstraintType.foreign_key,
+                columns=["a"],
+                expression="(a) REFERENCES `c`.`s`.`parent`",
+            ),
+            template_bundle.relation.identifier,
+        )
+        assert self._macro_name(template_bundle, constraint, model) == py_name
+
+    def test_parity__primary_key_single_column(self, template_bundle, model):
+        constraint = {"type": "primary_key", "columns": ["a"]}
+        py_name = synthesize_constraint_name(
+            PrimaryKeyConstraint(type=ConstraintType.primary_key, columns=["a"]),
+            template_bundle.relation.identifier,
+        )
+        assert self._macro_name(template_bundle, constraint, model) == py_name
+
+    def test_parity__primary_key_multiple_columns(self, template_bundle, model):
+        constraint = {"type": "primary_key", "columns": ["a", "b"]}
+        py_name = synthesize_constraint_name(
+            PrimaryKeyConstraint(type=ConstraintType.primary_key, columns=["a", "b"]),
+            template_bundle.relation.identifier,
+        )
+        assert self._macro_name(template_bundle, constraint, model) == py_name
+
+    def test_parity__primary_key_with_rely_expression(self, template_bundle, model):
+        constraint = {"type": "primary_key", "columns": ["a"], "expression": "RELY"}
+        py_name = synthesize_constraint_name(
+            PrimaryKeyConstraint(type=ConstraintType.primary_key, columns=["a"], expression="RELY"),
+            template_bundle.relation.identifier,
+        )
+        assert self._macro_name(template_bundle, constraint, model) == py_name