Fix multiple foreign key constraints bug

Author: ericj-db

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

@@ -71,7 +71,7 @@
   WHERE kcu.table_catalog = '{{ relation.database|lower }}'
     AND kcu.table_schema = '{{ relation.schema|lower }}'
     AND kcu.table_name = '{{ relation.identifier|lower }}'
-    AND kcu.constraint_name = (
+    AND kcu.constraint_name IN (
       SELECT constraint_name
       FROM `{{ relation.database|lower }}`.information_schema.table_constraints
       WHERE table_catalog = '{{ relation.database|lower }}'

tests/functional/adapter/incremental/fixtures.py

@@ -791,6 +791,16 @@ def model(dbt, spark):
     'parent' as type
 """
 
+fk_referenced_to_table_2 = """
+{{ config(
+    materialized = 'incremental',
+) }}
+
+select
+    cast(1 as bigint) as id,
+    'hello' as name
+"""
+
 constraint_schema_without_fk_constraint = """
 version: 2
 
@@ -814,7 +824,7 @@ def model(dbt, spark):
         data_type: string
 """
 
-constraint_schema_with_fk_constraint = """
+constraint_schema_with_fk_constraints = """
 version: 2
 
 models:
@@ -835,6 +845,19 @@ def model(dbt, spark):
       - name: type
         data_type: string
 
+  - name: fk_referenced_to_table_2
+    constraints:
+      - type: primary_key
+        columns: [id]
+        name: pk_parent_2
+    columns:
+      - name: id
+        data_type: bigint
+        constraints:
+          - type: not_null
+      - name: name
+        data_type: string
+
   - name: fk_referenced_from_table
     columns:
       - name: id
@@ -849,4 +872,9 @@ def model(dbt, spark):
         columns: [id, version]
         to: ref('fk_referenced_to_table')
         to_columns: [id, version]
+      - type: foreign_key
+        name: fk_to_parent_2
+        columns: [id]
+        to: ref('fk_referenced_to_table_2')
+        to_columns: [id]
 """

tests/functional/adapter/incremental/test_incremental_constraints.py

@@ -194,6 +194,7 @@ def project_config_update(self):
     def models(self):
         return {
             "fk_referenced_to_table.sql": fixtures.fk_referenced_to_table,
+            "fk_referenced_to_table_2.sql": fixtures.fk_referenced_to_table_2,
             "fk_referenced_from_table.sql": fixtures.fk_referenced_from_table,
             "schema.yml": fixtures.constraint_schema_without_fk_constraint,
         }
@@ -205,12 +206,14 @@ def test_add_foreign_key_constraint(self, project):
 
         # Foreign key constraint is informational only, so we cannot verify enforcement.
         # Instead, check that the metadata is updated correctly.
-        util.write_file(fixtures.constraint_schema_with_fk_constraint, "models", "schema.yml")
+        util.write_file(fixtures.constraint_schema_with_fk_constraints, "models", "schema.yml")
         util.run_dbt(["run"])
         referential_constraints = project.run_sql(referential_constraint_sql, fetch="all")
-        assert len(referential_constraints) == 1
+        assert len(referential_constraints) == 2
         assert referential_constraints[0][0] == "fk_to_parent"
         assert referential_constraints[0][1] == "pk_parent"
+        assert referential_constraints[1][0] == "fk_to_parent_2"
+        assert referential_constraints[1][1] == "pk_parent_2"
 
 
 @pytest.mark.skip_profile("databricks_cluster")
@@ -225,8 +228,9 @@ def project_config_update(self):
     def models(self):
         return {
             "fk_referenced_to_table.sql": fixtures.fk_referenced_to_table,
+            "fk_referenced_to_table_2.sql": fixtures.fk_referenced_to_table_2,
             "fk_referenced_from_table.sql": fixtures.fk_referenced_from_table,
-            "schema.yml": fixtures.constraint_schema_with_fk_constraint,
+            "schema.yml": fixtures.constraint_schema_with_fk_constraints,
         }
 
     def test_remove_foreign_key_constraint(self, project):
@@ -235,9 +239,11 @@ def test_remove_foreign_key_constraint(self, project):
 
         # Verify the constraint exists
         referential_constraints = project.run_sql(referential_constraint_sql, fetch="all")
-        assert len(referential_constraints) == 1
+        assert len(referential_constraints) == 2
         assert referential_constraints[0][0] == "fk_to_parent"
         assert referential_constraints[0][1] == "pk_parent"
+        assert referential_constraints[1][0] == "fk_to_parent_2"
+        assert referential_constraints[1][1] == "pk_parent_2"
 
         # Remove foreign key constraint and verify
         util.write_file(fixtures.constraint_schema_without_fk_constraint, "models", "schema.yml")