Repository: databricks/dbt-databricks
Pull request #1013 (2025-05-09 00:02:15)

Constraints for incremental materialization
<!-- Please review our pull request review process in CONTRIBUTING.md before your proceed. -->

Resolves constraint part of #982. Does not resolve the contract enforcement 

<!---
  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

Support constraints for incremental materializations. Databricks supports the following 4 constraints based on https://docs.databricks.com/aws/en/tables/constraints
- Non-null (enforced, requires UC)
- Check (enforced, does not require UC)
- Primary key (not enforced, requires UC)
- Foreign key (not enforced, requires UC)

### Checklist

- [ ] I have run this code in development and it appears to resolve the stated issue
- [ ] This PR includes tests, or tests are not required/relevant for this PR
- [ ] I have updated the `CHANGELOG.md` and added information about my change to the "dbt-databricks next" section.



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5598015e..77227116 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,9 @@
 ## dbt-databricks 1.10.2 (TBD)
 
+### Features
+
+- Support constraint updates on incremental runs (with Materialization V2) ([1013](https://github.com/databricks/dbt-databricks/pull/1013))
+
 ### Fixes
 
 - Fix bug with multiple not_null constraints defined on the model level ([1008](https://github.com/databricks/dbt-databricks/pull/1008))
diff --git a/dbt/adapters/databricks/constraints.py b/dbt/adapters/databricks/constraints.py
index 104dadcb..8ecded5b 100644
--- a/dbt/adapters/databricks/constraints.py
+++ b/dbt/adapters/databricks/constraints.py
@@ -1,4 +1,5 @@
 from abc import ABC, abstractmethod
+from dataclasses import dataclass
 from typing import Any, ClassVar, Optional, TypeVar
 from uuid import uuid4
 
@@ -27,6 +28,7 @@ CONSTRAINT_SUPPORT = {
 T = TypeVar("T", bound="TypedConstraint")
 
 
+@dataclass
 class TypedConstraint(ModelLevelConstraint, ABC):
     """Constraint that enforces type because it has render logic"""
 
@@ -61,6 +63,19 @@ class TypedConstraint(ModelLevelConstraint, ABC):
             f"{self.str_type} constraint '{name}' is missing required field(s): {fields}"
         )
 
+    # Enables set equality checks, especially for convenient unit testing
+    def __hash__(self) -> int:
+        # Create a tuple of all the fields that should be used for equality comparison
+        fields = (
+            self.type,
+            self.name,
+            tuple(self.columns) if self.columns else None,
+            self.expression,
+            self.to,
+            tuple(self.to_columns) if self.to_columns else None,
+        )
+        return hash(fields)
+
 
 class CustomConstraint(TypedConstraint):
     str_type = "custom"
diff --git a/dbt/adapters/databricks/impl.py b/dbt/adapters/databricks/impl.py
index 6185c625..074540b3 100644
--- a/dbt/adapters/databricks/impl.py
+++ b/dbt/adapters/databricks/impl.py
@@ -944,6 +944,15 @@ class IncrementalTableAPI(RelationAPIBase[IncrementalTableConfig]):
 
         if not relation.is_hive_metastore():
             results["information_schema.tags"] = adapter.execute_macro("fetch_tags", kwargs=kwargs)
+            results["non_null_constraint_columns"] = adapter.execute_macro(
+                "fetch_non_null_constraint_columns", kwargs=kwargs
+            )
+            results["primary_key_constraints"] = adapter.execute_macro(
+                "fetch_primary_key_constraints", kwargs=kwargs
+            )
+            results["foreign_key_constraints"] = adapter.execute_macro(
+                "fetch_foreign_key_constraints", kwargs=kwargs
+            )
         results["show_tblproperties"] = adapter.execute_macro("fetch_tbl_properties", kwargs=kwargs)
 
         kwargs = {"table_name": relation}
diff --git a/dbt/adapters/databricks/relation_configs/constraints.py b/dbt/adapters/databricks/relation_configs/constraints.py
new file mode 100644
index 00000000..f94bbcc4
--- /dev/null
+++ b/dbt/adapters/databricks/relation_configs/constraints.py
@@ -0,0 +1,175 @@
+from dataclasses import asdict
+from typing import ClassVar, Optional
+
+from agate import Table
+
+from dbt.adapters.contracts.relation import RelationConfig
+from dbt.adapters.databricks.constraints import (
+    CheckConstraint,
+    ConstraintType,
+    ForeignKeyConstraint,
+    PrimaryKeyConstraint,
+    TypedConstraint,
+    parse_constraints,
+)
+from dbt.adapters.databricks.relation_configs.base import (
+    DatabricksComponentConfig,
+    DatabricksComponentProcessor,
+)
+from dbt.adapters.relation_configs.config_base import RelationResults
+
+
+class ConstraintsConfig(DatabricksComponentConfig):
+    """Component encapsulating the constraints of a relation."""
+
+    set_non_nulls: set[str]
+    unset_non_nulls: set[str] = set()
+    set_constraints: set[TypedConstraint]
+    unset_constraints: set[TypedConstraint] = set()
+
+    def get_diff(self, other: "ConstraintsConfig") -> Optional["ConstraintsConfig"]:
+        # Find constraints that need to be unset
+        constraints_to_unset = other.set_constraints - self.set_constraints
+        # Find non-nulls that need to be unset
+        non_nulls_to_unset = other.set_non_nulls - self.set_non_nulls
+
+        # Set constraints that exist in self but not in other
+        set_constraints = self.set_constraints - other.set_constraints
+        set_non_nulls = self.set_non_nulls - other.set_non_nulls
+
+        if set_constraints or set_non_nulls or constraints_to_unset or non_nulls_to_unset:
+            return ConstraintsConfig(
+                set_non_nulls=set_non_nulls,
+                unset_non_nulls=non_nulls_to_unset,
+                set_constraints=set_constraints,
+                unset_constraints=constraints_to_unset,
+            )
+        return None
+
+
+class ConstraintsProcessor(DatabricksComponentProcessor[ConstraintsConfig]):
+    name: ClassVar[str] = "constraints"
+
+    @classmethod
+    def _process_check_constraints(
+        cls, check_constraints_table: Optional[Table]
+    ) -> set[CheckConstraint]:
+        check_constraints = set()
+        if check_constraints_table:
+            for row in check_constraints_table.rows:
+                if row[0].startswith("delta.constraints."):
+                    check_constraints.add(
+                        CheckConstraint(
+                            type=ConstraintType.check,
+                            name=row[0].replace("delta.constraints.", ""),
+                            expression=row[1],
+                        )
+                    )
+        return check_constraints
+
+    @classmethod
+    def _process_primary_key_constraints(
+        cls, pk_constraints_table: Optional[Table]
+    ) -> set[PrimaryKeyConstraint]:
+        pk_constraints = set()
+        if pk_constraints_table:
+            # Group columns by constraint name
+            constraint_columns: dict[str, list[str]] = {}
+            for row in pk_constraints_table.rows:
+                constraint_name = row[0]
+                column_name = row[1]
+                if constraint_name not in constraint_columns:
+                    constraint_columns[constraint_name] = []
+                constraint_columns[constraint_name].append(column_name)
+
+            # Create one PrimaryKeyConstraint per unique constraint name
+            for constraint_name, columns in constraint_columns.items():
+                pk_constraints.add(
+                    PrimaryKeyConstraint(
+                        type=ConstraintType.primary_key, name=constraint_name, columns=columns
+                    )
+                )
+        return pk_constraints
+
+    @classmethod
+    def _process_foreign_key_constraints(
+        cls, fk_constraints_table: Optional[Table]
+    ) -> set[ForeignKeyConstraint]:
+        fk_constraints = set()
+        if fk_constraints_table:
+            # Group by constraint_name
+            fk_data: dict[str, dict] = {}
+            for row in fk_constraints_table.rows:
+                constraint_name = row[0]
+                if constraint_name not in fk_data:
+                    fk_data[constraint_name] = {
+                        "columns": [],
+                        "to": f"`{row[2]}`.`{row[3]}`.`{row[4]}`",
+                        "to_columns": [],
+                    }
+                fk_data[constraint_name]["columns"].append(row[1])
+                fk_data[constraint_name]["to_columns"].append(row[5])
+
+            # Create ForeignKeyConstraint objects
+            for constraint_name, data in fk_data.items():
+                fk_constraints.add(
+                    ForeignKeyConstraint(
+                        type=ConstraintType.foreign_key,
+                        name=constraint_name,
+                        columns=data["columns"],
+                        to=data["to"],
+                        to_columns=data["to_columns"],
+                    )
+                )
+        return fk_constraints
+
+    @classmethod
+    def from_relation_results(cls, results: RelationResults) -> ConstraintsConfig:
+        non_null_constraint_columns = results.get("non_null_constraint_columns")
+        to_set_non_nulls = (
+            {row[0] for row in non_null_constraint_columns.rows}
+            if non_null_constraint_columns
+            else set()
+        )
+
+        table_properties = results.get("show_tblproperties")
+        primary_key_constraints = results.get("primary_key_constraints")
+        foreign_key_constraints = results.get("foreign_key_constraints")
+
+        check_constraints = cls._process_check_constraints(table_properties)
+        pk_constraints = cls._process_primary_key_constraints(primary_key_constraints)
+        fk_constraints = cls._process_foreign_key_constraints(foreign_key_constraints)
+
+        all_constraints: list[TypedConstraint] = []
+        all_constraints.extend(check_constraints)
+        all_constraints.extend(pk_constraints)
+        all_constraints.extend(fk_constraints)
+
+        return ConstraintsConfig(
+            set_non_nulls=to_set_non_nulls,
+            set_constraints=set(all_constraints),
+        )
+
+    @classmethod
+    def from_relation_config(cls, relation_config: RelationConfig) -> ConstraintsConfig:
+        constraints = getattr(relation_config, "constraints", [])
+        constraints = [
+            {
+                "name": (item["name"] if isinstance(item, dict) else item.name),
+                **(item if isinstance(item, dict) else asdict(item)),
+            }
+            for item in constraints
+        ]
+
+        columns = getattr(relation_config, "columns", {})
+        columns = [
+            {"name": name, **(col if isinstance(col, dict) else asdict(col))}
+            for name, col in columns.items()
+        ]
+
+        non_nulls, other_constraints = parse_constraints(columns, constraints)
+
+        return ConstraintsConfig(
+            set_non_nulls=set(non_nulls),
+            set_constraints=set(other_constraints),
+        )
diff --git a/dbt/adapters/databricks/relation_configs/incremental.py b/dbt/adapters/databricks/relation_configs/incremental.py
index e069e397..c549dd2b 100644
--- a/dbt/adapters/databricks/relation_configs/incremental.py
+++ b/dbt/adapters/databricks/relation_configs/incremental.py
@@ -3,6 +3,7 @@ from dbt.adapters.databricks.relation_configs.base import (
 )
 from dbt.adapters.databricks.relation_configs.column_comments import ColumnCommentsProcessor
 from dbt.adapters.databricks.relation_configs.comment import CommentProcessor
+from dbt.adapters.databricks.relation_configs.constraints import ConstraintsProcessor
 from dbt.adapters.databricks.relation_configs.liquid_clustering import LiquidClusteringProcessor
 from dbt.adapters.databricks.relation_configs.tags import TagsProcessor
 from dbt.adapters.databricks.relation_configs.tblproperties import TblPropertiesProcessor
@@ -12,6 +13,7 @@ class IncrementalTableConfig(DatabricksRelationConfigBase):
     config_components = [
         CommentProcessor,
         ColumnCommentsProcessor,
+        ConstraintsProcessor,
         TagsProcessor,
         TblPropertiesProcessor,
         LiquidClusteringProcessor,
diff --git a/dbt/include/databricks/macros/relations/components/constraints.sql b/dbt/include/databricks/macros/relations/components/constraints.sql
new file mode 100644
index 00000000..6b3b71f9
--- /dev/null
+++ b/dbt/include/databricks/macros/relations/components/constraints.sql
@@ -0,0 +1,135 @@
+{% macro fetch_non_null_constraint_columns(relation) -%}
+  {% if relation.is_hive_metastore() %}
+    {{ exceptions.raise_compiler_error("Incremental application of constraints is not supported for Hive Metastore") }}
+  {%- endif %}
+  {% call statement('list_non_null_constraint_columns', fetch_result=True) -%}
+    {{ fetch_non_null_constraint_columns_sql(relation) }}
+  {% endcall %}
+  {% do return(load_result('list_non_null_constraint_columns').table) %}
+{%- endmacro -%}
+
+{% macro fetch_non_null_constraint_columns_sql(relation) -%}
+  SELECT column_name
+  FROM `{{ relation.database|lower }}`.`information_schema`.`columns`
+  WHERE table_catalog = '{{ relation.database|lower }}' 
+    AND table_schema = '{{ relation.schema|lower }}'
+    AND table_name = '{{ relation.identifier|lower }}'
+    AND is_nullable = 'NO';
+{%- endmacro -%}
+
+{% macro fetch_primary_key_constraints(relation) -%}
+  {% if relation.is_hive_metastore() %}
+    {{ exceptions.raise_compiler_error("Incremental application of constraints is not supported for Hive Metastore") }}
+  {%- endif %}
+  {% call statement('list_primary_key_constraints', fetch_result=True) -%}
+    {{ fetch_primary_key_constraints_sql(relation) }}
+  {% endcall %}
+  {% do return(load_result('list_primary_key_constraints').table) %}
+{%- endmacro -%}
+
+{% macro fetch_primary_key_constraints_sql(relation) -%}
+  SELECT kcu.constraint_name, kcu.column_name
+  FROM `{{ relation.database|lower }}`.information_schema.key_column_usage kcu
+  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 = (
+      SELECT constraint_name
+      FROM `{{ relation.database|lower }}`.information_schema.table_constraints
+      WHERE table_catalog = '{{ relation.database|lower }}'
+        AND table_schema = '{{ relation.schema|lower }}'
+        AND table_name = '{{ relation.identifier|lower }}' 
+        AND constraint_type = 'PRIMARY KEY'
+    )
+  ORDER BY kcu.ordinal_position;
+{%- endmacro -%}
+
+{% macro fetch_foreign_key_constraints(relation) -%}
+  {% if relation.is_hive_metastore() %}
+    {{ exceptions.raise_compiler_error("Incremental application of constraints is not supported for Hive Metastore") }}
+  {%- endif %}
+  {% call statement('list_foreign_key_constraints', fetch_result=True) -%}
+    {{ fetch_foreign_key_constraints_sql(relation) }}
+  {% endcall %}
+  {% do return(load_result('list_foreign_key_constraints').table) %}
+{%- endmacro -%}
+
+{% macro fetch_foreign_key_constraints_sql(relation) -%}
+  SELECT
+    kcu.constraint_name,
+    kcu.column_name AS from_column,
+    ukcu.table_catalog AS to_catalog,
+    ukcu.table_schema AS to_schema,
+    ukcu.table_name AS to_table,
+    ukcu.column_name AS to_column
+  FROM `{{ relation.database|lower }}`.information_schema.key_column_usage kcu
+  JOIN `{{ relation.database|lower }}`.information_schema.referential_constraints rc
+    ON kcu.constraint_name = rc.constraint_name
+  JOIN `{{ relation.database|lower }}`.information_schema.key_column_usage ukcu
+    ON rc.unique_constraint_name = ukcu.constraint_name
+    AND kcu.ordinal_position = ukcu.ordinal_position
+  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 = (
+      SELECT constraint_name
+      FROM `{{ relation.database|lower }}`.information_schema.table_constraints
+      WHERE table_catalog = '{{ relation.database|lower }}'
+        AND table_schema = '{{ relation.schema|lower }}'
+        AND table_name = '{{ relation.identifier|lower }}'
+        AND constraint_type = 'FOREIGN KEY'
+    )
+  ORDER BY kcu.ordinal_position;
+{%- endmacro -%}
+
+{% macro apply_constraints(relation, constraints) -%}
+  {{ log("Applying constraints to relation " ~ constraints) }}
+  {%- if constraints and relation.is_hive_metastore() -%}
+    {{ exceptions.raise_compiler_error("Constraints are only supported for Unity Catalog") }}
+  {%- endif -%}
+  {# Order matters here because key constraints depend on non-null constraints #} 
+  {%- if constraints.unset_constraints %}
+    {%- for constraint in constraints.unset_constraints -%}
+      {%- call statement('main') -%}
+        {{ alter_unset_constraint(relation, constraint) }}
+      {%- endcall -%}
+    {%- endfor -%}
+  {%- endif %}
+  {%- if constraints.unset_non_nulls %}
+    {%- for column in constraints.unset_non_nulls -%}
+      {%- call statement('main') -%}
+        {{ alter_unset_non_null_constraint(relation, column) }}
+      {%- endcall -%}
+    {%- endfor -%}
+  {%- endif %}
+  {%- if constraints.set_non_nulls %}
+    {%- for column in constraints.set_non_nulls -%}
+      {%- call statement('main') -%}
+        {{ alter_set_non_null_constraint(relation, column) }}
+      {%- endcall -%}
+    {%- endfor -%}
+  {%- endif %}
+  {%- if constraints.set_constraints %}
+    {%- for constraint in constraints.set_constraints -%}
+      {%- call statement('main') -%}
+        {{ alter_set_constraint(relation, constraint) }}
+      {%- endcall -%}
+    {%- endfor -%}
+  {%- endif %}
+{%- endmacro -%}
+
+{% macro alter_set_non_null_constraint(relation, column) -%}
+  ALTER {{ relation.type }} {{ relation.render() }} ALTER COLUMN {{ column }} SET NOT NULL;
+{%- endmacro -%}
+
+{% macro alter_unset_non_null_constraint(relation, column) -%}
+  ALTER {{ relation.type }} {{ relation.render() }} ALTER COLUMN {{ column }} DROP NOT NULL;
+{%- endmacro -%}
+
+{% macro alter_set_constraint(relation, constraint) -%}
+  ALTER {{ relation.type }} {{ relation.render() }} ADD {{ constraint.render() }};
+{%- endmacro -%}
+
+{% macro alter_unset_constraint(relation, constraint) -%}
+  ALTER {{ relation.type }} {{ relation.render() }} DROP CONSTRAINT {{ constraint.name }};
+{%- endmacro -%}
diff --git a/dbt/include/databricks/macros/relations/table/alter.sql b/dbt/include/databricks/macros/relations/table/alter.sql
index 92899fa9..5ada83c8 100644
--- a/dbt/include/databricks/macros/relations/table/alter.sql
+++ b/dbt/include/databricks/macros/relations/table/alter.sql
@@ -6,6 +6,7 @@
       {% set tags = configuration_changes.changes.get("tags") %}
       {% set tblproperties = configuration_changes.changes.get("tblproperties") %}
       {% 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, tags.unset_tags) %}
       {%- endif -%}
@@ -21,5 +22,8 @@
       {% if column_comments %}
         {{ alter_column_comments(target_relation, column_comments.comments) }}
       {% endif %}
+      {% if constraints %}
+        {{ apply_constraints(target_relation, constraints) }}
+      {% endif %}
     {%- endif -%}
 {% endmacro %}
\ No newline at end of file


Test patch:
diff --git a/tests/functional/adapter/incremental/fixtures.py b/tests/functional/adapter/incremental/fixtures.py
index 18fcee40..a872f64f 100644
--- a/tests/functional/adapter/incremental/fixtures.py
+++ b/tests/functional/adapter/incremental/fixtures.py
@@ -620,3 +620,232 @@ replace_expected = """id,msg,color
 1,hello,blue
 2,goodbye,red
 """
+
+non_null_constraint_sql = """
+{{ config(
+    materialized = 'incremental',
+) }}
+
+{% if not is_incremental() %}
+
+select cast(1 as bigint) as id, 'hello' as msg
+
+{% else %}
+
+select cast(2 as bigint) as id, cast(NULL as string) as msg
+
+{% endif %}
+"""
+
+
+schema_without_non_null_constraint = """
+version: 2
+
+models:
+  - name: non_null_constraint_sql
+    columns:
+      - name: id
+        data_type: bigint
+      - name: msg
+        data_type: string
+"""
+
+schema_with_non_null_constraint = """
+version: 2
+
+models:
+  - name: non_null_constraint_sql
+    columns:
+      - name: id
+        data_type: bigint
+      - name: msg
+        data_type: string
+        constraints:
+          - type: not_null
+"""
+
+check_constraint_sql = """
+{{ config(
+    materialized = 'incremental',
+) }}
+
+{% if not is_incremental() %}
+
+select cast(6 as bigint) as id
+
+{% else %}
+
+select cast(3 as bigint) as id
+
+{% endif %}
+"""
+
+schema_without_check_constraint = """
+version: 2
+
+models:
+  - name: check_constraint_sql
+    columns:
+      - name: id
+        data_type: bigint
+"""
+
+schema_with_check_constraint = """
+version: 2
+
+models:
+  - name: check_constraint_sql
+    columns:
+      - name: id
+        data_type: bigint
+    constraints:
+      - type: check
+        name: id_greater_than_5
+        expression: id > 5
+"""
+
+primary_key_constraint_sql = """
+{{ config(
+    materialized = 'incremental',
+) }}
+
+select
+    cast(1 as bigint) as id,
+    cast(1 as int) as version,
+    'hello' as msg
+"""
+
+schema_with_single_column_primary_key_constraint = """
+version: 2
+
+models:
+  - name: primary_key_constraint_sql
+    columns:
+      - name: id
+        data_type: bigint
+        constraints:
+          - type: not_null
+      - name: version
+        data_type: int
+      - name: msg
+        data_type: string
+    constraints:
+      - type: primary_key
+        name: pk_model
+        columns: [id]
+"""
+
+schema_with_composite_primary_key_constraint = """
+version: 2
+
+models:
+  - name: primary_key_constraint_sql
+    columns:
+      - name: id
+        data_type: bigint
+        constraints:
+          - type: not_null
+      - name: version
+        data_type: int
+        constraints:
+          - type: not_null
+      - name: msg
+        data_type: string
+    constraints:
+      - type: primary_key
+        name: pk_model_updated
+        columns: [id, version]
+"""
+
+fk_referenced_from_table = """
+{{ config(
+    materialized = 'incremental',
+) }}
+
+{% if not is_incremental() %}
+
+select
+    cast(1 as bigint) as id,
+    cast(1 as int) as version,
+    'hello' as msg
+
+{% else %}
+
+select
+    cast(2 as bigint) as id,
+    cast(2 as int) as version,
+    'world' as msg
+
+{% endif %}
+"""
+
+fk_referenced_to_table = """
+{{ config(
+    materialized = 'incremental',
+) }}
+
+select
+    cast(1 as bigint) as id,
+    cast(1 as int) as version,
+    'parent' as type
+"""
+
+constraint_schema_without_fk_constraint = """
+version: 2
+
+models:
+  - name: fk_referenced_to_table
+    columns:
+      - name: id
+        data_type: bigint
+      - name: version
+        data_type: int
+      - name: type
+        data_type: string
+
+  - name: fk_referenced_from_table
+    columns:
+      - name: id
+        data_type: bigint
+      - name: version
+        data_type: int
+      - name: msg
+        data_type: string
+"""
+
+constraint_schema_with_fk_constraint = """
+version: 2
+
+models:
+  - name: fk_referenced_to_table
+    constraints:
+      - type: primary_key
+        columns: [id, version]
+        name: pk_parent
+    columns:
+      - name: id
+        data_type: bigint
+        constraints:
+          - type: not_null
+      - name: version
+        data_type: int
+        constraints:
+          - type: not_null
+      - name: type
+        data_type: string
+
+  - name: fk_referenced_from_table
+    columns:
+      - name: id
+        data_type: bigint
+      - name: version
+        data_type: int
+      - name: msg
+        data_type: string
+    constraints:
+      - type: foreign_key
+        name: fk_to_parent
+        columns: [id, version]
+        to: ref('fk_referenced_to_table')
+        to_columns: [id, version]
+"""
diff --git a/tests/functional/adapter/incremental/test_incremental_constraints.py b/tests/functional/adapter/incremental/test_incremental_constraints.py
new file mode 100644
index 00000000..76c5df82
--- /dev/null
+++ b/tests/functional/adapter/incremental/test_incremental_constraints.py
@@ -0,0 +1,246 @@
+import pytest
+
+from dbt.contracts.results import RunStatus
+from dbt.tests import util
+from tests.functional.adapter.incremental import fixtures
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestIncrementalSetNonNullConstraint:
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {"use_materialization_v2": True},
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "non_null_constraint_sql.sql": fixtures.non_null_constraint_sql,
+            "schema.yml": fixtures.schema_without_non_null_constraint,
+        }
+
+    def test_add_non_null_constraint(self, project):
+        results = util.run_dbt(["run"], expect_pass=True)
+        util.write_file(fixtures.schema_with_non_null_constraint, "models", "schema.yml")
+
+        # Non-null constraint is an enforced constraint, so this materialization should fail
+        results = util.run_dbt(["run"], expect_pass=False)
+        assert results.results[0].status == RunStatus.Error
+        assert "DELTA_NOT_NULL_CONSTRAINT_VIOLATED" in results.results[0].message
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestIncrementalUnsetNonNullConstraint:
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {"use_materialization_v2": True},
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "non_null_constraint_sql.sql": fixtures.non_null_constraint_sql,
+            "schema.yml": fixtures.schema_with_non_null_constraint,
+        }
+
+    def test_remove_non_null_constraint(self, project):
+        util.run_dbt(["run"])
+        # Verify the constraint exists
+        columns = project.run_sql(
+            """
+            SELECT column_name
+            FROM {database}.information_schema.columns
+            WHERE table_schema = '{schema}'
+            AND is_nullable = 'NO'
+            """,
+            fetch="all",
+        )
+        assert len(columns) == 1
+        assert columns[0][0] == "msg"
+
+        # Remove the non-null constraint
+        util.write_file(fixtures.schema_without_non_null_constraint, "models", "schema.yml")
+        # This would fail if the constraint was not removed
+        util.run_dbt(["run"])
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestIncrementalSetCheckConstraint:
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {"use_materialization_v2": True},
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "check_constraint_sql.sql": fixtures.check_constraint_sql,
+            "schema.yml": fixtures.schema_without_check_constraint,
+        }
+
+    def test_add_check_constraint(self, project):
+        results = util.run_dbt(["run"])
+        util.write_file(fixtures.schema_with_check_constraint, "models", "schema.yml")
+
+        # Check constraint is an enforced constraint, so this materialization should fail
+        results = util.run_dbt(["run"], expect_pass=False)
+        assert results.results[0].status == RunStatus.Error
+        assert "CHECK constraint" in results.results[0].message
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestIncrementalRemoveCheckConstraint:
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {"use_materialization_v2": True},
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "check_constraint_sql.sql": fixtures.check_constraint_sql,
+            "schema.yml": fixtures.schema_with_check_constraint,
+        }
+
+    def test_remove_check_constraint(self, project):
+        # First run with check constraint
+        util.run_dbt(["run"])
+
+        # Verify the constraint exists
+        tbl_properties = project.run_sql(
+            """
+            SHOW TBLPROPERTIES {database}.{schema}.check_constraint_sql
+            """,
+            fetch="all",
+        )
+        constraint = None
+        for row in tbl_properties:
+            if str(row[0]).startswith("delta.constraints."):
+                constraint = row
+                break
+        assert constraint is not None
+
+        # Remove check constraint
+        util.write_file(fixtures.schema_without_check_constraint, "models", "schema.yml")
+        # This would fail if the constraint was not removed
+        util.run_dbt(["run"])
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestIncrementalUpdatePrimaryKeyConstraint:
+    primary_key_constraint_sql = """
+        SELECT constraint_name, column_name
+        FROM {database}.information_schema.key_column_usage
+        WHERE constraint_schema = '{schema}'
+        ORDER BY ordinal_position
+    """
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {"use_materialization_v2": True},
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "primary_key_constraint_sql.sql": fixtures.primary_key_constraint_sql,
+            "schema.yml": fixtures.schema_with_single_column_primary_key_constraint,
+        }
+
+    def test_update_primary_key_constraint(self, project):
+        # First run with single column primary key
+        util.run_dbt(["run"])
+        primary_key_constraints = project.run_sql(self.primary_key_constraint_sql, fetch="all")
+        assert len(primary_key_constraints) == 1
+        assert primary_key_constraints[0][0] == "pk_model"
+        assert primary_key_constraints[0][1] == "id"
+
+        # Update to composite key. Under the hood, it will trigger both remove/add operations
+        util.write_file(
+            fixtures.schema_with_composite_primary_key_constraint, "models", "schema.yml"
+        )
+        util.run_dbt(["run"])
+        primary_key_constraints = project.run_sql(self.primary_key_constraint_sql, fetch="all")
+        assert len(primary_key_constraints) == 2
+        assert primary_key_constraints[0][0] == "pk_model_updated"
+        assert primary_key_constraints[0][1] == "id"
+        assert primary_key_constraints[1][0] == "pk_model_updated"
+        assert primary_key_constraints[1][1] == "version"
+        # Verify previous constraint was removed
+        assert not any(constraint[0] == "pk_model" for constraint in primary_key_constraints)
+
+
+referential_constraint_sql = """
+    SELECT constraint_name, unique_constraint_name
+    FROM {database}.information_schema.referential_constraints
+    WHERE constraint_schema = '{schema}'
+"""
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestIncrementalSetForeignKeyConstraint:
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {"use_materialization_v2": True},
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "fk_referenced_to_table.sql": fixtures.fk_referenced_to_table,
+            "fk_referenced_from_table.sql": fixtures.fk_referenced_from_table,
+            "schema.yml": fixtures.constraint_schema_without_fk_constraint,
+        }
+
+    def test_add_foreign_key_constraint(self, project):
+        util.run_dbt(["run"])
+        referential_constraints = project.run_sql(referential_constraint_sql, fetch="all")
+        assert len(referential_constraints) == 0
+
+        # 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.run_dbt(["run"])
+        referential_constraints = project.run_sql(referential_constraint_sql, fetch="all")
+        assert len(referential_constraints) == 1
+        assert referential_constraints[0][0] == "fk_to_parent"
+        assert referential_constraints[0][1] == "pk_parent"
+
+
+@pytest.mark.skip_profile("databricks_cluster")
+class TestIncrementalRemoveForeignKeyConstraint:
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {"use_materialization_v2": True},
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "fk_referenced_to_table.sql": fixtures.fk_referenced_to_table,
+            "fk_referenced_from_table.sql": fixtures.fk_referenced_from_table,
+            "schema.yml": fixtures.constraint_schema_with_fk_constraint,
+        }
+
+    def test_remove_foreign_key_constraint(self, project):
+        # First run with foreign key constraint
+        util.run_dbt(["run"])
+
+        # Verify the constraint exists
+        referential_constraints = project.run_sql(referential_constraint_sql, fetch="all")
+        assert len(referential_constraints) == 1
+        assert referential_constraints[0][0] == "fk_to_parent"
+        assert referential_constraints[0][1] == "pk_parent"
+
+        # Remove foreign key constraint and verify
+        util.write_file(fixtures.constraint_schema_without_fk_constraint, "models", "schema.yml")
+        util.run_dbt(["run"])
+        referential_constraints = project.run_sql(referential_constraint_sql, fetch="all")
+        assert len(referential_constraints) == 0
diff --git a/tests/unit/relation_configs/test_constraint.py b/tests/unit/relation_configs/test_constraint.py
new file mode 100644
index 00000000..9247f81e
--- /dev/null
+++ b/tests/unit/relation_configs/test_constraint.py
@@ -0,0 +1,325 @@
+from unittest.mock import Mock
+
+from agate import Table
+from dbt_common.contracts.constraints import (
+    ColumnLevelConstraint,
+    ConstraintType,
+    ModelLevelConstraint,
+)
+
+from dbt.adapters.databricks.constraints import (
+    CheckConstraint,
+    ForeignKeyConstraint,
+    PrimaryKeyConstraint,
+)
+from dbt.adapters.databricks.relation_configs.constraints import (
+    ConstraintsConfig,
+    ConstraintsProcessor,
+)
+from dbt.artifacts.resources.v1.components import ColumnInfo
+from tests.unit import fixtures
+
+
+class TestConstraintsProcessor:
+    def test_from_relation_results__none(self):
+        results = {
+            "non_null_constraint_columns": None,
+            "show_tblproperties": None,
+            "primary_key_constraints": None,
+            "foreign_key_constraints": None,
+        }
+        spec = ConstraintsProcessor.from_relation_results(results)
+        assert spec == ConstraintsConfig(set_non_nulls=set(), set_constraints=set())
+
+    def test_from_relation_results__with_check_constraints(self):
+        results = {
+            "show_tblproperties": fixtures.gen_tblproperties(
+                [
+                    ["delta.constraints.check_name_length", "LENGTH (name) >= 1"],
+                    ["other_prop", "value"],
+                ]
+            ),
+        }
+        spec = ConstraintsProcessor.from_relation_results(results)
+        assert spec == ConstraintsConfig(
+            set_non_nulls=set(),
+            set_constraints={
+                CheckConstraint(
+                    type=ConstraintType.check,
+                    name="check_name_length",
+                    expression="LENGTH (name) >= 1",
+                )
+            },
+        )
+
+    def test_from_relation_results__with_primary_key_constraints(self):
+        results = {
+            "primary_key_constraints": Table(
+                rows=[
+                    ["pk_user", "id"],
+                    ["pk_user", "email"],
+                ],
+                column_names=["constraint_name", "column_name"],
+            ),
+        }
+        spec = ConstraintsProcessor.from_relation_results(results)
+        assert spec == ConstraintsConfig(
+            set_non_nulls=set(),
+            set_constraints={
+                PrimaryKeyConstraint(
+                    type=ConstraintType.primary_key,
+                    name="pk_user",
+                    columns=["id", "email"],
+                )
+            },
+        )
+
+    def test_from_relation_results__with_foreign_key_constraints(self):
+        results = {
+            "foreign_key_constraints": Table(
+                rows=[
+                    ["fk_user_1", "id", "catalog", "schema", "customers", "customer_id"],
+                    ["fk_user_1", "email", "catalog", "schema", "customers", "email"],
+                ],
+                column_names=[
+                    "constraint_name",
+                    "from_column",
+                    "to_catalog",
+                    "to_schema",
+                    "to_table",
+                    "to_column",
+                ],
+            ),
+        }
+        spec = ConstraintsProcessor.from_relation_results(results)
+        assert spec == ConstraintsConfig(
+            set_non_nulls=set(),
+            set_constraints={
+                ForeignKeyConstraint(
+                    type=ConstraintType.foreign_key,
+                    name="fk_user_1",
+                    columns=["id", "email"],
+                    to="`catalog`.`schema`.`customers`",
+                    to_columns=["customer_id", "email"],
+                )
+            },
+        )
+
+    def test_from_relation_results__with_non_null_constraints(self):
+        results = {
+            "non_null_constraint_columns": Table(
+                rows=[
+                    ["id"],
+                    ["email"],
+                ],
+                column_names=["column_name"],
+            ),
+        }
+        spec = ConstraintsProcessor.from_relation_results(results)
+        assert spec == ConstraintsConfig(
+            set_non_nulls={"id", "email"},
+            set_constraints=set(),
+        )
+
+    def test_from_relation_config__without_constraints(self):
+        model = Mock()
+        model.constraints = []
+        model.columns = {}
+        spec = ConstraintsProcessor.from_relation_config(model)
+        assert spec == ConstraintsConfig(set_non_nulls=set(), set_constraints=set())
+
+    def test_from_relation_config__with_check_constraint(self):
+        model = Mock()
+        model.columns = {}
+        model.constraints = [
+            ModelLevelConstraint(
+                type=ConstraintType.check,
+                name="check_name_length",
+                expression="LENGTH (name) >= 1",
+            )
+        ]
+        spec = ConstraintsProcessor.from_relation_config(model)
+        assert spec == ConstraintsConfig(
+            set_non_nulls=set(),
+            set_constraints={
+                CheckConstraint(
+                    type=ConstraintType.check,
+                    name="check_name_length",
+                    expression="(LENGTH (name) >= 1)",
+                )
+            },
+        )
+
+    def test_from_relation_config__with_primary_key_constraint(self):
+        model = Mock()
+        model.columns = {}
+        model.constraints = [
+            ModelLevelConstraint(
+                type=ConstraintType.primary_key,
+                name="pk_user",
+                columns=["id", "email"],
+            )
+        ]
+        spec = ConstraintsProcessor.from_relation_config(model)
+        assert spec == ConstraintsConfig(
+            set_non_nulls=set(),
+            set_constraints={
+                PrimaryKeyConstraint(
+                    type=ConstraintType.primary_key,
+                    name="pk_user",
+                    columns=["id", "email"],
+                )
+            },
+        )
+
+    def test_from_relation_config__with_foreign_key_constraint(self):
+        model = Mock()
+        model.columns = {}
+        model.constraints = [
+            ModelLevelConstraint(
+                type=ConstraintType.foreign_key,
+                name="fk_user_customer",
+                columns=["id", "email"],
+                to="`catalog`.`schema`.`customers`",
+                to_columns=["customer_id", "email"],
+            )
+        ]
+        spec = ConstraintsProcessor.from_relation_config(model)
+        assert spec == ConstraintsConfig(
+            set_non_nulls=set(),
+            set_constraints={
+                ForeignKeyConstraint(
+                    type=ConstraintType.foreign_key,
+                    name="fk_user_customer",
+                    columns=["id", "email"],
+                    to="`catalog`.`schema`.`customers`",
+                    to_columns=["customer_id", "email"],
+                )
+            },
+        )
+
+    def test_from_relation_config__with_non_null_constraint(self):
+        model = Mock()
+        model.columns = {
+            "email": ColumnInfo(
+                name="email",
+                constraints=[ColumnLevelConstraint(type=ConstraintType.not_null)],
+            ),
+        }
+        model.constraints = [
+            ModelLevelConstraint(
+                type=ConstraintType.not_null,
+                columns=["id"],
+            )
+        ]
+        spec = ConstraintsProcessor.from_relation_config(model)
+
+        assert spec == ConstraintsConfig(
+            set_non_nulls={"email", "id"},
+            set_constraints=set(),
+        )
+
+
+class TestConstraintsConfig:
+    def test_get_diff__empty_and_some_exist(self):
+        config = ConstraintsConfig(set_non_nulls=set(), set_constraints=set())
+        other = ConstraintsConfig(
+            set_non_nulls={"id"},
+            set_constraints={
+                PrimaryKeyConstraint(
+                    type=ConstraintType.primary_key,
+                    name="user_pk",
+                    columns=["id"],
+                )
+            },
+        )
+        diff = config.get_diff(other)
+        assert diff == ConstraintsConfig(
+            set_non_nulls=set(),
+            unset_non_nulls={"id"},
+            set_constraints=set(),
+            unset_constraints={
+                PrimaryKeyConstraint(
+                    type=ConstraintType.primary_key,
+                    name="user_pk",
+                    columns=["id"],
+                )
+            },
+        )
+
+    def test_get_diff__some_new_and_empty_existing(self):
+        config = ConstraintsConfig(
+            set_non_nulls={"id"},
+            set_constraints={
+                PrimaryKeyConstraint(
+                    type=ConstraintType.primary_key,
+                    name="pk_user",
+                    columns=["id"],
+                )
+            },
+        )
+        other = ConstraintsConfig(set_non_nulls=set(), set_constraints=set())
+        diff = config.get_diff(other)
+        assert diff == ConstraintsConfig(
+            set_non_nulls={"id"},
+            unset_non_nulls=set(),
+            set_constraints={
+                PrimaryKeyConstraint(
+                    type=ConstraintType.primary_key,
+                    name="pk_user",
+                    columns=["id"],
+                )
+            },
+            unset_constraints=set(),
+        )
+
+    def test_get_diff__mixed_case(self):
+        config = ConstraintsConfig(
+            set_non_nulls={"id", "email"},
+            set_constraints={
+                CheckConstraint(
+                    type=ConstraintType.check,
+                    name="check_name_length",
+                    expression="LENGTH (name) >= 1",
+                ),
+                PrimaryKeyConstraint(
+                    type=ConstraintType.primary_key,
+                    name="pk_user",
+                    columns=["id"],
+                ),
+            },
+        )
+        other = ConstraintsConfig(
+            set_non_nulls={"id", "name"},
+            set_constraints={
+                CheckConstraint(
+                    type=ConstraintType.check,
+                    name="check_name_length",
+                    expression="LENGTH (name) >= 1",
+                ),
+                PrimaryKeyConstraint(
+                    type=ConstraintType.primary_key,
+                    name="pk_user",
+                    columns=["name"],
+                ),
+            },
+        )
+        diff = config.get_diff(other)
+        assert diff == ConstraintsConfig(
+            set_non_nulls={"email"},
+            unset_non_nulls={"name"},
+            set_constraints={
+                PrimaryKeyConstraint(
+                    type=ConstraintType.primary_key,
+                    name="pk_user",
+                    columns=["id"],
+                )
+            },
+            unset_constraints={
+                PrimaryKeyConstraint(
+                    type=ConstraintType.primary_key,
+                    name="pk_user",
+                    columns=["name"],
+                )
+            },
+        )
diff --git a/tests/unit/relation_configs/test_incremental_config.py b/tests/unit/relation_configs/test_incremental_config.py
index 907f8ebc..a66ef522 100644
--- a/tests/unit/relation_configs/test_incremental_config.py
+++ b/tests/unit/relation_configs/test_incremental_config.py
@@ -2,6 +2,13 @@ from agate import Table
 
 from dbt.adapters.databricks.relation_configs.column_comments import ColumnCommentsConfig
 from dbt.adapters.databricks.relation_configs.comment import CommentConfig
+from dbt.adapters.databricks.relation_configs.constraints import (
+    CheckConstraint,
+    ConstraintsConfig,
+    ConstraintType,
+    ForeignKeyConstraint,
+    PrimaryKeyConstraint,
+)
 from dbt.adapters.databricks.relation_configs.incremental import IncrementalTableConfig
 from dbt.adapters.databricks.relation_configs.liquid_clustering import LiquidClusteringConfig
 from dbt.adapters.databricks.relation_configs.tags import TagsConfig
@@ -23,6 +30,7 @@ class TestIncrementalConfig:
                     ["prop", "f1"],
                     ["clusterByAuto", "true"],
                     ["clusteringColumns", '[["col1"],[""a""]]'],
+                    ["delta.constraints.check_name_length", "LENGTH (name) >= 1"],
                 ],
                 column_names=["key", "value"],
             ),
@@ -32,6 +40,35 @@ class TestIncrementalConfig:
                 ],
                 column_names=["col_name", "col_type", "comment"],
             ),
+            "non_null_constraint_columns": Table(
+                rows=[
+                    ["id"],
+                    ["email"],
+                ],
+                column_names=["column_name"],
+            ),
+            "primary_key_constraints": Table(
+                rows=[
+                    ["pk_user", "id"],
+                    ["pk_user", "email"],
+                ],
+                column_names=["constraint_name", "column_name"],
+            ),
+            "foreign_key_constraints": Table(
+                rows=[
+                    ["fk_user_1", "id", "catalog", "schema", "customers", "customer_id"],
+                    ["fk_user_1", "email", "catalog", "schema", "customers", "email"],
+                    ["fk_user_2", "id", "catalog", "schema", "employees", "employee_id"],
+                ],
+                column_names=[
+                    "constraint_name",
+                    "from_column",
+                    "to_catalog",
+                    "to_schema",
+                    "to_table",
+                    "to_column",
+                ],
+            ),
         }
 
         config = IncrementalTableConfig.from_results(results)
@@ -43,10 +80,46 @@ class TestIncrementalConfig:
                 "column_comments": ColumnCommentsConfig(
                     comments={"column": "test comment"}, quoted={}, persist=False
                 ),
-                "tblproperties": TblPropertiesConfig(tblproperties={"prop": "f1"}),
+                "tblproperties": TblPropertiesConfig(
+                    tblproperties={
+                        "prop": "f1",
+                        "delta.constraints.check_name_length": "LENGTH (name) >= 1",
+                    }
+                ),
                 "liquid_clustering": LiquidClusteringConfig(
                     auto_cluster=True,
                     cluster_by=["col1", '"a"'],
                 ),
+                "constraints": ConstraintsConfig(
+                    set_non_nulls=["id", "email"],
+                    unset_non_nulls=[],
+                    set_constraints=[
+                        CheckConstraint(
+                            type=ConstraintType.check,
+                            name="check_name_length",
+                            expression="LENGTH (name) >= 1",
+                        ),
+                        PrimaryKeyConstraint(
+                            type=ConstraintType.primary_key,
+                            name="pk_user",
+                            columns=["id", "email"],
+                        ),
+                        ForeignKeyConstraint(
+                            type=ConstraintType.foreign_key,
+                            name="fk_user_1",
+                            to="`catalog`.`schema`.`customers`",
+                            to_columns=["customer_id", "email"],
+                            columns=["id", "email"],
+                        ),
+                        ForeignKeyConstraint(
+                            type=ConstraintType.foreign_key,
+                            name="fk_user_2",
+                            to="`catalog`.`schema`.`employees`",
+                            to_columns=["employee_id"],
+                            columns=["id"],
+                        ),
+                    ],
+                    unset_constraints=[],
+                ),
             }
         )