Repository: databricks/dbt-databricks
Pull request #876 (2024-12-12 23:06:21)

Redo constraints (since columns can only take one constraint at create time)
<!---
  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

When I went to apply the constraint logic from the previous PR, I discovered that Databricks does not like receiving multiple constraints for a single column in the table create syntax; however, you can add arbitrarily many table level constraints, so in this PR I aim to do the following:

* if a column has a not_null constraint, directly set not_null on the column
* any other column constraint gets converted to a TypedConstraint, a subclass of ModelLevelConstraint that enforces the requirements of the particular constraint, and holds the rendering logic.  The conversion required setting 'columns=[name]'. Because python makes operator overloading a pain in the butt, I'm going back to OOP for the render logic.
* If a model level constraint is not_null, convert to setting not_null on each affected column, since Databricks does not support not_null constraint definition as a table constraint
* any other model constraint gets converted to a TypedConstraint

### Checklist

- [x] I have run this code in development and it appears to resolve the stated issue
- [x] 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 01d3db19..f8fa84a4 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,20 +6,9 @@
 
 ### Under the Hood
 
-- Implement new constraint logic for use_materialization_v2 flag ([846](https://github.com/databricks/dbt-databricks/pull/846/files))
+- Implement new constraint logic for use_materialization_v2 flag ([846](https://github.com/databricks/dbt-databricks/pull/846/files)), ([876](https://github.com/databricks/dbt-databricks/pull/876))
 
-## dbt-databricks 1.9.1 (TBD)
-
-### Features
-
-- Merge strategy now supports the `update set ...` action with the explicit list of updates for `when not matched by source` ([866](https://github.com/databricks/dbt-databricks/pull/866)) (thanks @mi-volodin).
-
-### Under the Hood
-
-- Removed pins for pandas and pydantic to ease user burdens ([874](https://github.com/databricks/dbt-databricks/pull/874))
-- Add more relation types to make codegen happy ([875](https://github.com/databricks/dbt-databricks/pull/875))
-
-## dbt-databricks 1.9.0 (December 9, 2024)
+## dbt-databricks 1.9.0 (Dec 9, 2024)
 
 ### Features
 
diff --git a/dbt/adapters/databricks/column.py b/dbt/adapters/databricks/column.py
index 4e15e912..7e718fbe 100644
--- a/dbt/adapters/databricks/column.py
+++ b/dbt/adapters/databricks/column.py
@@ -1,9 +1,6 @@
 from dataclasses import dataclass
 from typing import Any, ClassVar, Optional
 
-from dbt_common.contracts.constraints import ColumnLevelConstraint, ConstraintType
-
-from dbt.adapters.databricks import constraints
 from dbt.adapters.databricks.utils import quote
 from dbt.adapters.spark.column import SparkColumn
 
@@ -13,7 +10,6 @@ class DatabricksColumn(SparkColumn):
     table_comment: Optional[str] = None
     comment: Optional[str] = None
     not_null: Optional[bool] = None
-    constraints: Optional[list[ColumnLevelConstraint]] = None
 
     TYPE_LABELS: ClassVar[dict[str, str]] = {
         "LONG": "BIGINT",
@@ -32,17 +28,7 @@ class DatabricksColumn(SparkColumn):
     def data_type(self) -> str:
         return self.translate_type(self.dtype)
 
-    def add_constraint(self, constraint: ColumnLevelConstraint) -> None:
-        # On first constraint add, initialize constraint details
-        if self.constraints is None:
-            self.constraints = []
-            self.not_null = False
-        if constraint.type == ConstraintType.not_null:
-            self.not_null = True
-        else:
-            self.constraints.append(constraint)
-
-    def enrich(self, model_column: dict[str, Any]) -> "DatabricksColumn":
+    def enrich(self, model_column: dict[str, Any], not_null: bool) -> "DatabricksColumn":
         """Create a copy that incorporates model column metadata, including constraints."""
 
         data_type = model_column.get("data_type") or self.dtype
@@ -50,11 +36,7 @@ class DatabricksColumn(SparkColumn):
         if model_column.get("description"):
             enriched_column.comment = model_column["description"]
 
-        if model_column.get("constraints"):
-            for constraint in model_column["constraints"]:
-                parsed_constraint = constraints.parse_column_constraint(constraint)
-                enriched_column.add_constraint(parsed_constraint)
-
+        enriched_column.not_null = not_null
         return enriched_column
 
     def render_for_create(self) -> str:
@@ -65,10 +47,6 @@ class DatabricksColumn(SparkColumn):
         if self.comment:
             comment = self.comment.replace("'", "\\'")
             column_str += f" COMMENT '{comment}'"
-        for constraint in self.constraints or []:
-            c = constraints.process_column_constraint(constraint)
-            if c:
-                column_str += f" {c}"
         return column_str
 
     def __repr__(self) -> str:
diff --git a/dbt/adapters/databricks/constraints.py b/dbt/adapters/databricks/constraints.py
index 96380090..5761dacf 100644
--- a/dbt/adapters/databricks/constraints.py
+++ b/dbt/adapters/databricks/constraints.py
@@ -1,5 +1,5 @@
-from functools import partial
-from typing import Any, Callable, Optional, TypeVar
+from abc import ABC, abstractmethod
+from typing import Any, ClassVar, Optional, Type, TypeVar
 
 from dbt_common.contracts.constraints import (
     ColumnLevelConstraint,
@@ -10,7 +10,6 @@ from dbt_common.events.functions import warn_or_error
 from dbt_common.exceptions import DbtValidationError
 
 from dbt.adapters.base import ConstraintSupport
-from dbt.adapters.databricks.logging import logger
 from dbt.adapters.events.types import ConstraintNotEnforced, ConstraintNotSupported
 
 # Support constants
@@ -22,21 +21,98 @@ CONSTRAINT_SUPPORT = {
     ConstraintType.foreign_key: ConstraintSupport.NOT_ENFORCED,
 }
 
-SUPPORTED_FOR_COLUMN = {
-    ConstraintType.custom,
-    ConstraintType.primary_key,
-    ConstraintType.foreign_key,
-}
-
 # Types
 """Generic type variable for constraints."""
-T = TypeVar("T", bound=ColumnLevelConstraint)
+T = TypeVar("T", bound="TypedConstraint")
+
+
+class TypedConstraint(ModelLevelConstraint, ABC):
+    """Constraint that enforces type because it has render logic"""
+
+    str_type: ClassVar[str]
+
+    @classmethod
+    def __post_deserialize__(cls: Type[T], obj: T) -> T:
+        assert obj.type == cls.str_type, "Mismatched constraint type"
+        obj._validate()
+        return obj
+
+    def _validate(self) -> None:
+        return
+
+    def render(self) -> str:
+        return f"{self._render_prefix()}{self._render_suffix()}"
+
+    def _render_prefix(self) -> str:
+        if self.name:
+            return f"CONSTRAINT {self.name} "
+        return ""
+
+    @abstractmethod
+    def _render_suffix(self) -> str:
+        pass
+
+    @classmethod
+    def _render_error(self, missing: list[list[str]]) -> DbtValidationError:
+        fields = " or ".join(["(" + ", ".join([f"'{e}'" for e in x]) + ")" for x in missing])
+        name = self.name or ""
+        return DbtValidationError(
+            f"{self.str_type} constraint '{name}' is missing required field(s): {fields}"
+        )
+
+
+class CustomConstraint(TypedConstraint):
+    str_type = "custom"
+
+    def _validate(self) -> None:
+        if self.expression is None:
+            raise self._render_error([["expression"]])
+
+    def _render_suffix(self) -> str:
+        return self.expression or ""
+
+
+class PrimaryKeyConstraint(TypedConstraint):
+    str_type = "primary_key"
+
+    def _validate(self) -> None:
+        if not self.columns:
+            raise self._render_error([["columns"]])
+
+    def _render_suffix(self) -> str:
+        suffix = f"PRIMARY KEY ({', '.join(self.columns)})"
+        if self.expression:
+            suffix += f" {self.expression}"
+        return suffix
+
 
-"""Function type for checking constraint support."""
-SupportFunc = Callable[[T], bool]
+class ForeignKeyConstraint(TypedConstraint):
+    str_type = "foreign_key"
 
-"""Function type for rendering constraints."""
-RenderFunc = Callable[[T], str]
+    def _validate(self) -> None:
+        if not self.columns or (not (self.to_columns and self.to) and not self.expression):
+            raise self._render_error(
+                [["columns", "to", "to_columns"], ["columns", "expression"]],
+            )
+
+    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
+
+
+class CheckConstraint(TypedConstraint):
+    str_type = "check"
+
+    def _validate(self) -> None:
+        if not self.expression:
+            raise self._render_error([["expression"]])
+
+    def _render_suffix(self) -> str:
+        return f"CHECK ({self.expression})"
 
 
 # Base support and enforcement
@@ -55,32 +131,18 @@ def is_enforced(constraint: ColumnLevelConstraint) -> bool:
     ]
 
 
-# Core parsing, validation, and processing
-def parse_constraint(klass: type[T], raw_constraint: dict[str, Any]) -> T:
-    try:
-        klass.validate(raw_constraint)
-        return klass.from_dict(raw_constraint)
-    except Exception:
-        raise DbtValidationError(f"Could not parse constraint: {raw_constraint}")
-
-
-def process_constraint(
-    constraint: T,
-    support_func: SupportFunc,
-    render_funcs: dict[ConstraintType, RenderFunc],
-) -> Optional[str]:
-    if validate_constraint(constraint, support_func):
-        return render_constraint(constraint, render_funcs)
-
+def process_constraint(constraint: TypedConstraint) -> Optional[str]:
+    if validate_constraint(constraint):
+        return constraint.render()
     return None
 
 
-def validate_constraint(constraint: T, support_func: SupportFunc) -> bool:
+def validate_constraint(constraint: ColumnLevelConstraint) -> bool:
     # Custom constraints are always supported
     if constraint.type == ConstraintType.custom:
         return True
 
-    supported = support_func(constraint)
+    supported = is_supported(constraint)
 
     if constraint.warn_unsupported and not supported:
         warn_or_error(
@@ -94,141 +156,65 @@ def validate_constraint(constraint: T, support_func: SupportFunc) -> bool:
     return supported
 
 
-def render_constraint(
-    constraint: T, render_funcs: dict[ConstraintType, RenderFunc]
-) -> Optional[str]:
-    rendered_constraint = ""
-
-    if constraint.type in render_funcs:
-        if constraint.name:
-            rendered_constraint = f"CONSTRAINT {constraint.name} "
-        rendered_constraint += render_funcs[constraint.type](constraint)
-
-    rendered_constraint = rendered_constraint.strip()
-
-    return rendered_constraint if rendered_constraint != "" else None
-
-
-def supported_for(constraint: T, support_func: SupportFunc, warning: str) -> bool:
-    if is_supported(constraint) and not support_func(constraint):
-        logger.warning(warning.format(type=constraint.type))
-
-    return is_supported(constraint) and support_func(constraint)
-
-
-# Shared render functions
-def render_error(constraint: ColumnLevelConstraint, missing: list[list[str]]) -> DbtValidationError:
-    fields = " or ".join(["(" + ", ".join([f"'{e}'" for e in x]) + ")" for x in missing])
-    constraint_type = constraint.type.value
-    constraint_name = "" if not constraint.name else f"{constraint.name} "
-    return DbtValidationError(
-        f"{constraint_type} constraint {constraint_name}is missing required field(s): {fields}"
-    )
-
-
-def render_custom(constraint: ColumnLevelConstraint) -> str:
-    assert constraint.type == ConstraintType.custom
-    if constraint.expression:
-        return constraint.expression
-    raise render_error(constraint, [["expression"]])
-
-
-# ColumnLevelConstraint specialization
-def parse_column_constraint(raw_constraint: dict[str, Any]) -> ColumnLevelConstraint:
-    return parse_constraint(ColumnLevelConstraint, raw_constraint)
-
-
-COLUMN_WARNING = (
-    "While constraint of type {type} is supported for models, it is not supported for columns."
-)
-
-supported_for_columns = partial(
-    supported_for, support_func=lambda x: x.type in SUPPORTED_FOR_COLUMN, warning=COLUMN_WARNING
-)
-
-
-def column_constraint_map() -> dict[ConstraintType, RenderFunc]:
-    return {
-        ConstraintType.primary_key: render_primary_key_for_column,
-        ConstraintType.foreign_key: render_foreign_key_for_column,
-        ConstraintType.custom: render_custom,
-    }
-
-
-def process_column_constraint(constraint: ColumnLevelConstraint) -> Optional[str]:
-    return process_constraint(constraint, supported_for_columns, column_constraint_map())
-
-
-def render_primary_key_for_column(constraint: ColumnLevelConstraint) -> str:
-    assert constraint.type == ConstraintType.primary_key
-    rendered = "PRIMARY KEY"
-    if constraint.expression:
-        rendered += f" {constraint.expression}"
-    return rendered
-
-
-def render_foreign_key_for_column(constraint: ColumnLevelConstraint) -> str:
-    assert constraint.type == ConstraintType.foreign_key
-    rendered = "FOREIGN KEY"
-    if constraint.expression:
-        return rendered + f" {constraint.expression}"
-    elif constraint.to and constraint.to_columns:
-        return rendered + f" REFERENCES {constraint.to} ({', '.join(constraint.to_columns)})"
-    raise render_error(constraint, [["expression"], ["to", "to_columns"]])
-
-
-# ModelLevelConstraint specialization
-def parse_model_constraint(raw_constraint: dict[str, Any]) -> ModelLevelConstraint:
-    return parse_constraint(ModelLevelConstraint, raw_constraint)
-
-
-MODEL_WARNING = (
-    "While constraint of type {type} is supported for columns, it is not supported for models."
-)
-
-
-supported_for_models = partial(
-    supported_for, support_func=lambda x: x.type != ConstraintType.not_null, warning=MODEL_WARNING
-)
-
-
-def model_constraint_map() -> dict[ConstraintType, RenderFunc]:
-    return {
-        ConstraintType.primary_key: render_primary_key_for_model,
-        ConstraintType.foreign_key: render_foreign_key_for_model,
-        ConstraintType.custom: render_custom,
-        ConstraintType.check: render_check,
-    }
-
-
-def process_model_constraint(constraint: ModelLevelConstraint) -> Optional[str]:
-    return process_constraint(constraint, supported_for_models, model_constraint_map())
-
-
-def render_primary_key_for_model(constraint: ModelLevelConstraint) -> str:
-    prefix = render_primary_key_for_column(constraint)
-    if constraint.expression:
-        return prefix
-    if constraint.columns:
-        return f"{prefix} ({', '.join(constraint.columns)})"
-    raise render_error(constraint, [["columns"], ["expression"]])
-
-
-def render_foreign_key_for_model(constraint: ModelLevelConstraint) -> str:
-    assert constraint.type == ConstraintType.foreign_key
-    rendered = "FOREIGN KEY"
-    if constraint.columns and constraint.to and constraint.to_columns:
-        columns = ", ".join(constraint.columns)
-        to_columns = ", ".join(constraint.to_columns)
-        return rendered + f" ({columns}) REFERENCES {constraint.to} ({to_columns})"
-    elif constraint.expression:
-        return rendered + " " + constraint.expression
-    raise render_error(constraint, [["expression"], ["columns", "to", "to_columns"]])
-
+def parse_constraints(
+    model_columns: list[dict[str, Any]], model_constraints: list[dict[str, Any]]
+) -> tuple[set[str], list[TypedConstraint]]:
+    not_nulls, constraints_from_columns = parse_column_constraints(model_columns)
+    not_nulls_from_models, constraints_from_models = parse_model_constraints(model_constraints)
+
+    return not_nulls.union(
+        not_nulls_from_models
+    ), constraints_from_columns + constraints_from_models
+
+
+def parse_column_constraints(
+    model_columns: list[dict[str, Any]],
+) -> tuple[set[str], list[TypedConstraint]]:
+    column_names: set[str] = set()
+    constraints: list[TypedConstraint] = []
+    for column in model_columns:
+        for constraint in column.get("constraints", []):
+            if constraint["type"] == ConstraintType.not_null:
+                column_names.add(column["name"])
+            else:
+                constraint["columns"] = [column["name"]]
+                constraints.append(parse_constraint(constraint))
+
+    return column_names, constraints
+
+
+def parse_model_constraints(
+    model_constraints: list[dict[str, Any]],
+) -> tuple[set[str], list[TypedConstraint]]:
+    column_names: set[str] = set()
+    constraints: list[TypedConstraint] = []
+    for constraint in model_constraints:
+        if constraint["type"] == ConstraintType.not_null:
+            if not constraint.get("columns"):
+                raise DbtValidationError("not_null constraint on model must have 'columns' defined")
+            column_names.add(*constraint["columns"])
+        else:
+            constraints.append(parse_constraint(constraint))
+
+    return column_names, constraints
+
+
+CONSTRAINT_TYPE_MAP = {
+    "primary_key": PrimaryKeyConstraint,
+    "foreign_key": ForeignKeyConstraint,
+    "check": CheckConstraint,
+    "custom": CustomConstraint,
+    "not_null": ColumnLevelConstraint,
+    "unique": ColumnLevelConstraint,
+}
 
-def render_check(constraint: ColumnLevelConstraint) -> str:
-    assert constraint.type == ConstraintType.check
-    if constraint.expression:
-        return f"CHECK ({constraint.expression})"
 
-    raise render_error(constraint, [["expression"]])
+def parse_constraint(
+    raw_constraint: dict[str, Any], type_map: dict[str, type[T]] = CONSTRAINT_TYPE_MAP
+) -> T:
+    try:
+        klass = type_map[raw_constraint["type"]]
+        klass.validate(raw_constraint)
+        return klass.from_dict(raw_constraint)
+    except Exception:
+        raise DbtValidationError(f"Could not parse constraint: {raw_constraint}")
diff --git a/dbt/adapters/databricks/impl.py b/dbt/adapters/databricks/impl.py
index 5e4b75d6..9a6f57ed 100644
--- a/dbt/adapters/databricks/impl.py
+++ b/dbt/adapters/databricks/impl.py
@@ -769,21 +769,28 @@ class DatabricksAdapter(SparkAdapter):
 
     @available
     @staticmethod
-    def get_enriched_columns(
-        existing_columns: list[DatabricksColumn], model_columns: dict[str, dict[str, Any]]
-    ) -> list[DatabricksColumn]:
+    def parse_columns_and_constraints(
+        existing_columns: list[DatabricksColumn],
+        model_columns: dict[str, dict[str, Any]],
+        model_constraints: list[dict[str, Any]],
+    ) -> tuple[list[DatabricksColumn], list[constraints.TypedConstraint]]:
         """Returns a list of columns that have been updated with features for table create."""
         enriched_columns = []
+        not_null_set, parsed_constraints = constraints.parse_constraints(
+            list(model_columns.values()), model_constraints
+        )
 
         for column in existing_columns:
             if column.name in model_columns:
                 column_info = model_columns[column.name]
-                enriched_column = column.enrich(column_info)
+                enriched_column = column.enrich(column_info, column.name in not_null_set)
                 enriched_columns.append(enriched_column)
             else:
+                if column.name in not_null_set:
+                    column.not_null = True
                 enriched_columns.append(column)
 
-        return enriched_columns
+        return enriched_columns, parsed_constraints
 
 
 @dataclass(frozen=True)
diff --git a/dbt/adapters/databricks/relation.py b/dbt/adapters/databricks/relation.py
index 1d66e359..efa091a4 100644
--- a/dbt/adapters/databricks/relation.py
+++ b/dbt/adapters/databricks/relation.py
@@ -2,7 +2,7 @@ from collections.abc import Iterable
 from dataclasses import dataclass, field
 from typing import Any, Optional, Type
 
-from dbt_common.contracts.constraints import ConstraintType, ModelLevelConstraint
+from dbt_common.contracts.constraints import ConstraintType
 from dbt_common.dataclass_schema import StrEnum
 from dbt_common.exceptions import DbtRuntimeError
 from dbt_common.utils import filter_null_values
@@ -11,7 +11,7 @@ from dbt.adapters.base.relation import BaseRelation, InformationSchema, Policy
 from dbt.adapters.contracts.relation import (
     ComponentName,
 )
-from dbt.adapters.databricks.constraints import parse_model_constraint, process_model_constraint
+from dbt.adapters.databricks.constraints import TypedConstraint, process_constraint
 from dbt.adapters.databricks.utils import remove_undefined
 from dbt.adapters.spark.impl import KEY_TABLE_OWNER, KEY_TABLE_STATISTICS
 from dbt.adapters.utils import classproperty
@@ -41,8 +41,6 @@ class DatabricksRelationType(StrEnum):
     Foreign = "foreign"
     StreamingTable = "streaming_table"
     External = "external"
-    ManagedShallowClone = "managed_shallow_clone"
-    ExternalShallowClone = "external_shallow_clone"
     Unknown = "unknown"
 
 
@@ -62,8 +60,8 @@ class DatabricksRelation(BaseRelation):
     quote_policy: Policy = field(default_factory=lambda: DatabricksQuotePolicy())
     include_policy: Policy = field(default_factory=lambda: DatabricksIncludePolicy())
     quote_character: str = "`"
-    create_constraints: list[ModelLevelConstraint] = field(default_factory=list)
-    alter_constraints: list[ModelLevelConstraint] = field(default_factory=list)
+    create_constraints: list[TypedConstraint] = field(default_factory=list)
+    alter_constraints: list[TypedConstraint] = field(default_factory=list)
     metadata: Optional[dict[str, Any]] = None
 
     @classmethod
@@ -153,21 +151,21 @@ class DatabricksRelation(BaseRelation):
     def StreamingTable(cls) -> str:
         return str(DatabricksRelationType.StreamingTable)
 
-    def add_constraint(self, constraint: ModelLevelConstraint) -> None:
+    def add_constraint(self, constraint: TypedConstraint) -> None:
         if constraint.type == ConstraintType.check:
             self.alter_constraints.append(constraint)
         else:
             self.create_constraints.append(constraint)
 
-    def enrich(self, raw_constraints: list[dict[str, Any]]) -> "DatabricksRelation":
+    def enrich(self, constraints: list[TypedConstraint]) -> "DatabricksRelation":
         copy = self.incorporate()
-        for constraint in raw_constraints:
-            copy.add_constraint(parse_model_constraint(constraint))
+        for constraint in constraints:
+            copy.add_constraint(constraint)
 
         return copy
 
     def render_constraints_for_create(self) -> str:
-        processed = [process_model_constraint(c) for c in self.create_constraints]
+        processed = map(process_constraint, self.create_constraints)
         return ", ".join(c for c in processed if c is not None)
 
 
diff --git a/dbt/include/databricks/macros/materializations/incremental/strategies.sql b/dbt/include/databricks/macros/materializations/incremental/strategies.sql
index 311c2f40..57db5496 100644
--- a/dbt/include/databricks/macros/materializations/incremental/strategies.sql
+++ b/dbt/include/databricks/macros/materializations/incremental/strategies.sql
@@ -91,15 +91,8 @@ select {{source_cols_csv}} from {{ source_relation }}
 
   {%- set not_matched_by_source_action = config.get('not_matched_by_source_action') -%}
   {%- set not_matched_by_source_condition = config.get('not_matched_by_source_condition') -%}
-
-  {%- set not_matched_by_source_action_trimmed = not_matched_by_source_action | lower | trim(' \n\t') %}
-  {%- set not_matched_by_source_action_is_set = (
-      not_matched_by_source_action_trimmed == 'delete'
-      or not_matched_by_source_action_trimmed.startswith('update')
-    )
-  %}
-  
   
+
   {% if unique_key %}
       {% if unique_key is sequence and unique_key is not mapping and unique_key is not string %}
           {% for key in unique_key %}
@@ -144,12 +137,12 @@ select {{source_cols_csv}} from {{ source_relation }}
         then insert
             {{ get_merge_insert(on_schema_change, source_columns, source_alias) }}
     {%- endif %}
-    {%- if not_matched_by_source_action_is_set %}
+    {%- if not_matched_by_source_action == 'delete' %}
     when not matched by source
         {%- if not_matched_by_source_condition %}
         and ({{ not_matched_by_source_condition }})
         {%- endif %}
-        then {{ not_matched_by_source_action }}
+        then delete
     {%- endif %}
 {% endmacro %}
 
diff --git a/docs/databricks-merge.md b/docs/databricks-merge.md
index 4034b8f6..caa00336 100644
--- a/docs/databricks-merge.md
+++ b/docs/databricks-merge.md
@@ -18,11 +18,7 @@ From v.1.9 onwards `merge` behavior can be tuned by setting the additional param
 
   - `skip_matched_step`: if set to `true`, dbt will completely skip the `matched` clause of the merge statement.
   - `skip_not_matched_step`: similarly if `true` the `not matched` clause will be skipped.
-  - `not_matched_by_source_action`: can be set to an action for the case the record does not exist in a source dataset. 
-    - if set to `delete` the corresponding `when not matched by source ... then delete` clause will be added to the merge statement. 
-    - if the action starts with `update` then the format `update set <actions>` is assumed, which will run update statement syntactically as provided.
-      Can be multiline formatted.
-    - in other cases by default no action is taken and now error raised.
+  - `not_matched_by_source_action`: if set to `delete` the corresponding `when not matched by source ... then delete` clause will be added to the merge statement.
   - `merge_with_schema_evolution`: when set to `true` dbt generates the merge statement with `WITH SCHEMA EVOLUTION` clause.
 
 - Step conditions that are expressed with an explicit SQL predicates allow to execute corresponding action only in case the conditions are met in addition to matching by the `unique_key`.
@@ -44,11 +40,7 @@ Example below illustrates how these parameters affect the merge statement genera
     matched_condition='t.tech_change_ts < s.tech_change_ts',
     not_matched_condition='s.attr1 IS NOT NULL',
     not_matched_by_source_condition='t.tech_change_ts < current_timestamp()',
-    not_matched_by_source_action='''
-        update set
-            t.attr1 = 'deleted',
-            t.tech_change_ts = current_timestamp()
-    ''',
+    not_matched_by_source_action='delete',
     merge_with_schema_evolution=true
 ) }}
 
@@ -101,7 +93,5 @@ when not matched
 
 when not matched by source
     and t.tech_change_ts < current_timestamp()
-    then update set
-        t.attr1 = 'deleted',
-        t.tech_change_ts = current_timestamp()
+    then delete
 ```
diff --git a/pyproject.toml b/pyproject.toml
index 55949b65..a21c8a99 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -29,7 +29,7 @@ dependencies = [
     "dbt-core>=1.9.0rc2, <2.0",
     "dbt-spark>=1.9.0b1, <2.0",
     "keyring>=23.13.0",
-    "pydantic>=1.10.0",
+    "pydantic>=1.10.0, <2",
 ]
 
 [project.urls]


Test patch:
diff --git a/tests/functional/adapter/incremental/fixtures.py b/tests/functional/adapter/incremental/fixtures.py
index 18fcee40..9d0f2913 100644
--- a/tests/functional/adapter/incremental/fixtures.py
+++ b/tests/functional/adapter/incremental/fixtures.py
@@ -264,14 +264,7 @@ matching_condition_expected = """id,first,second,V
 4,Baron,Harkonnen,1
 """
 
-not_matched_by_source_then_del_expected = """id,first,second,V
-2,Paul,Atreides,0
-3,Dunkan,Aidaho,1
-4,Baron,Harkonnen,1
-"""
-
-not_matched_by_source_then_upd_expected = """id,first,second,V
-1,--,--,-1
+not_matched_by_source_expected = """id,first,second,V
 2,Paul,Atreides,0
 3,Dunkan,Aidaho,1
 4,Baron,Harkonnen,1
@@ -418,7 +411,7 @@ select 5 as id, 'Raban' as first, '' as second, 0 as V -- no append
 {% endif %}
 """
 
-not_matched_by_source_then_delete_model = """
+not_matched_by_source_model = """
 {{ config(
     materialized = 'incremental',
     unique_key = 'id',
@@ -453,46 +446,6 @@ select 4 as id, 'Baron' as first, 'Harkonnen' as second, 1 as V -- should append
 {% endif %}
 """
 
-not_matched_by_source_then_update_model = """
-{{ config(
-    materialized = 'incremental',
-    unique_key = 'id',
-    incremental_strategy='merge',
-    target_alias='t',
-    source_alias='s',
-    skip_matched_step=true,
-    not_matched_by_source_condition='t.V > 0',
-    not_matched_by_source_action='''
-        update set
-            t.first = \\\'--\\\',
-            t.second = \\\'--\\\',
-            t.V = -1
-    ''',
-) }}
-
-{% if not is_incremental() %}
-
--- data for first invocation of model
-
-select 1 as id, 'Vasya' as first, 'Pupkin' as second, 1 as V
-union all
-select 2 as id, 'Paul' as first, 'Atreides' as second, 0 as V
-union all
-select 3 as id, 'Dunkan' as first, 'Aidaho' as second, 1 as V
-
-{% else %}
-
--- data for subsequent incremental update
-
--- id = 1 should be updated with
--- id = 2 should be kept as condition doesn't hold (t.V = 0)
-select 3 as id, 'Dunkan' as first, 'Aidaho' as second, 2 as V -- No update, skipped
-union all
-select 4 as id, 'Baron' as first, 'Harkonnen' as second, 1 as V -- should append
-
-{% endif %}
-"""
-
 merge_schema_evolution_model = """
 {{ config(
     materialized = 'incremental',
diff --git a/tests/functional/adapter/incremental/test_incremental_strategies.py b/tests/functional/adapter/incremental/test_incremental_strategies.py
index 45a24362..88db60ee 100644
--- a/tests/functional/adapter/incremental/test_incremental_strategies.py
+++ b/tests/functional/adapter/incremental/test_incremental_strategies.py
@@ -288,38 +288,17 @@ class TestMatchedAndNotMatchedCondition(IncrementalBase):
         )
 
 
-class TestNotMatchedBySourceAndConditionThenDelete(IncrementalBase):
+class TestNotMatchedBySourceAndCondition(IncrementalBase):
     @pytest.fixture(scope="class")
     def seeds(self):
         return {
-            "not_matched_by_source_expected.csv": fixtures.not_matched_by_source_then_del_expected,
+            "not_matched_by_source_expected.csv": fixtures.not_matched_by_source_expected,
         }
 
     @pytest.fixture(scope="class")
     def models(self):
         return {
-            "not_matched_by_source.sql": fixtures.not_matched_by_source_then_delete_model,
-        }
-
-    def test_merge(self, project):
-        self.seed_and_run_twice()
-        util.check_relations_equal(
-            project.adapter,
-            ["not_matched_by_source", "not_matched_by_source_expected"],
-        )
-
-
-class TestNotMatchedBySourceAndConditionThenUpdate(IncrementalBase):
-    @pytest.fixture(scope="class")
-    def seeds(self):
-        return {
-            "not_matched_by_source_expected.csv": fixtures.not_matched_by_source_then_upd_expected,
-        }
-
-    @pytest.fixture(scope="class")
-    def models(self):
-        return {
-            "not_matched_by_source.sql": fixtures.not_matched_by_source_then_update_model,
+            "not_matched_by_source.sql": fixtures.not_matched_by_source_model,
         }
 
     def test_merge(self, project):
diff --git a/tests/unit/test_adapter.py b/tests/unit/test_adapter.py
index 1eb4585b..6bd95860 100644
--- a/tests/unit/test_adapter.py
+++ b/tests/unit/test_adapter.py
@@ -427,7 +427,6 @@ class TestDatabricksAdapter(DatabricksAdapterBase):
             "numeric_precision": None,
             "char_size": None,
             "comment": None,
-            "constraints": None,
             "not_null": None,
         }
 
diff --git a/tests/unit/test_column.py b/tests/unit/test_column.py
index 57ea44fd..95fbbee5 100644
--- a/tests/unit/test_column.py
+++ b/tests/unit/test_column.py
@@ -1,5 +1,4 @@
 import pytest
-from dbt_common.contracts.constraints import ColumnLevelConstraint, ConstraintType
 
 from dbt.adapters.databricks.column import DatabricksColumn
 
@@ -29,23 +28,6 @@ class TestSparkColumn:
         }
 
 
-class TestAddConstraint:
-    @pytest.fixture
-    def column(self):
-        return DatabricksColumn("id", "LONG")
-
-    def test_add_constraint__not_null(self, column):
-        column.add_constraint(ColumnLevelConstraint(type=ConstraintType.not_null))
-        assert column.not_null is True
-        assert column.constraints == []
-
-    def test_add_constraint__other_constraint(self, column):
-        constraint = ColumnLevelConstraint(type=ConstraintType.custom)
-        column.add_constraint(constraint)
-        assert column.not_null is False
-        assert column.constraints == [constraint]
-
-
 class TestEnrich:
     @pytest.fixture
     def column(self):
@@ -63,13 +45,10 @@ class TestEnrich:
         }
 
     def test_enrich(self, column, model_column):
-        enriched_column = column.enrich(model_column)
+        enriched_column = column.enrich(model_column, True)
         assert enriched_column.data_type == "bigint"
         assert enriched_column.comment == "this is a column"
         assert enriched_column.not_null is True
-        assert enriched_column.constraints == [
-            ColumnLevelConstraint(type=ConstraintType.primary_key, name="foo")
-        ]
 
 
 class TestRenderForCreate:
@@ -88,23 +67,6 @@ class TestRenderForCreate:
         column.comment = "this is a column"
         assert column.render_for_create() == "id INT COMMENT 'this is a column'"
 
-    def test_render_for_create__constraints(self, column):
-        column.constraints = [
-            ColumnLevelConstraint(type=ConstraintType.primary_key),
-        ]
-        assert column.render_for_create() == "id INT PRIMARY KEY"
-
-    def test_render_for_create__everything(self, column):
-        column.not_null = True
-        column.comment = "this is a column"
-        column.constraints = [
-            ColumnLevelConstraint(type=ConstraintType.primary_key),
-            ColumnLevelConstraint(type=ConstraintType.custom, expression="foo"),
-        ]
-        assert column.render_for_create() == (
-            "id INT NOT NULL COMMENT 'this is a column' " "PRIMARY KEY foo"
-        )
-
     def test_render_for_create__escaping(self, column):
         column.comment = "this is a 'column'"
         assert column.render_for_create() == "id INT COMMENT 'this is a \\'column\\''"
diff --git a/tests/unit/test_constraints.py b/tests/unit/test_constraints.py
index 0b6fc593..b49fed8f 100644
--- a/tests/unit/test_constraints.py
+++ b/tests/unit/test_constraints.py
@@ -4,11 +4,139 @@ import pytest
 from dbt_common.contracts.constraints import (
     ColumnLevelConstraint,
     ConstraintType,
-    ModelLevelConstraint,
 )
 from dbt_common.exceptions import DbtValidationError
 
-from dbt.adapters.databricks import constraints
+from dbt.adapters.databricks.constraints import (
+    CheckConstraint,
+    CustomConstraint,
+    ForeignKeyConstraint,
+    PrimaryKeyConstraint,
+    TypedConstraint,
+    is_enforced,
+    is_supported,
+    parse_column_constraints,
+    parse_constraint,
+    parse_constraints,
+    parse_model_constraints,
+    process_constraint,
+    validate_constraint,
+)
+
+
+class TestConstraint(TypedConstraint):
+    str_type = "unique"
+
+    def _render_suffix(self):
+        return "test"
+
+
+class TestTypedConstraint:
+    def test_typed_constraint_from_dict__invalid_type(self):
+        with pytest.raises(AssertionError, match="Mismatched constraint type"):
+            TestConstraint.from_dict({"type": "custom"})
+
+    def test_render__with_name(self):
+        constraint = TestConstraint(type=ConstraintType.check, name="my_constraint")
+        assert constraint.render() == "CONSTRAINT my_constraint test"
+
+
+class TestCustomConstraint:
+    def test_custom_constraint_from_dict__valid(self):
+        raw = {"type": "custom", "expression": "1 = 1"}
+        assert CustomConstraint(
+            type=ConstraintType.custom, expression="1 = 1"
+        ) == CustomConstraint.from_dict(raw)
+
+    def test_custom_constraint_from_dict__invalid(self):
+        raw = {"type": "custom"}
+        with pytest.raises(
+            DbtValidationError, match="custom constraint '' is missing required field"
+        ):
+            CustomConstraint.from_dict(raw)
+
+    def test_custom_constraint_render(self):
+        constraint = CustomConstraint(type=ConstraintType.custom, expression="1 = 1")
+        assert constraint.render() == "1 = 1"
+
+
+class TestCheckConstraint:
+    def test_check_constraint_from_dict__valid(self):
+        raw = {"type": "check", "expression": "1 = 1"}
+        assert CheckConstraint(
+            type=ConstraintType.check, expression="1 = 1"
+        ) == CheckConstraint.from_dict(raw)
+
+    def test_check_constraint_validate__invalid(self):
+        raw = {"type": "check"}
+        with pytest.raises(
+            DbtValidationError, match="check constraint '' is missing required field"
+        ):
+            CheckConstraint.from_dict(raw)
+
+    def test_custom_constraint_render(self):
+        constraint = CheckConstraint(type=ConstraintType.check, expression="1 = 1")
+        assert constraint.render() == "CHECK (1 = 1)"
+
+
+class TestPrimaryKeyConstraint:
+    def test_primary_key_constraint_from_dict__valid(self):
+        raw = {"type": "primary_key", "columns": ["id"]}
+        assert PrimaryKeyConstraint(
+            type=ConstraintType.primary_key, columns=["id"]
+        ) == PrimaryKeyConstraint.from_dict(raw)
+
+    def test_primary_key_constraint_validate__invalid(self):
+        raw = {"type": "primary_key"}
+        with pytest.raises(
+            DbtValidationError, match="primary_key constraint '' is missing required field"
+        ):
+            PrimaryKeyConstraint.from_dict(raw)
+
+    def test_primary_key_constraint_render__no_expression(self):
+        constraint = PrimaryKeyConstraint(type=ConstraintType.primary_key, columns=["id"])
+        assert constraint.render() == "PRIMARY KEY (id)"
+
+    def test_primary_key_constraint_render__with_expression(self):
+        constraint = PrimaryKeyConstraint(
+            type=ConstraintType.primary_key, columns=["id", "other"], expression="DEFERRABLE"
+        )
+        assert constraint.render() == "PRIMARY KEY (id, other) DEFERRABLE"
+
+
+class TestForeignKeyConstraint:
+    def test_foreign_key_constraint_from_dict__valid(self):
+        raw = {"type": "foreign_key", "columns": ["id"], "expression": "1 = 1"}
+        assert ForeignKeyConstraint(
+            type=ConstraintType.foreign_key, columns=["id"], expression="1 = 1"
+        ) == ForeignKeyConstraint.from_dict(raw)
+
+    def test_foreign_key_constraint_validate__invalid(self):
+        raw = {"type": "foreign_key"}
+        with pytest.raises(
+            DbtValidationError, match="foreign_key constraint '' is missing required field"
+        ):
+            ForeignKeyConstraint.from_dict(raw)
+
+    def test_foreign_key_constraint_render__to(self):
+        constraint = ForeignKeyConstraint(
+            type=ConstraintType.foreign_key,
+            columns=["id"],
+            to="other_table",
+            to_columns=["other_id"],
+        )
+        assert constraint.render() == "FOREIGN KEY (id) REFERENCES other_table (other_id)"
+
+    def test_foreign_key_constraint_render__with_expression(self):
+        constraint = ForeignKeyConstraint(
+            type=ConstraintType.foreign_key,
+            columns=["id", "other"],
+            expression="REFERENCES other_table (other_id) DEFERRABLE",
+        )
+        assert (
+            constraint.render()
+            == "FOREIGN KEY (id, other) REFERENCES other_table (other_id) DEFERRABLE"
+        )
 
 
 class TestConstraintsSupported:
@@ -26,7 +154,7 @@ class TestConstraintsSupported:
     )
     def test_supported__expected(self, constraint_type, supported):
         constraint = ColumnLevelConstraint(type=constraint_type)
-        assert constraints.is_supported(constraint) == supported
+        assert is_supported(constraint) == supported
 
 
 class TestConstraintsEnforced:
@@ -44,43 +172,46 @@ class TestConstraintsEnforced:
     )
     def test_enforced__expected(self, constraint_type, enforced):
         constraint = ColumnLevelConstraint(type=constraint_type)
-        assert constraints.is_enforced(constraint) == enforced
+        assert is_enforced(constraint) == enforced
 
 
 class TestParseConstraint:
-    @pytest.mark.parametrize("klass", [ColumnLevelConstraint, ModelLevelConstraint])
-    def test_parse_constraint__valid_column(self, klass):
-        raw_constraint = {"type": "not_null"}
-        constraint = constraints.parse_constraint(klass, raw_constraint)
-        assert isinstance(constraint, klass)
-        assert constraint.type == ConstraintType.not_null
-
-    @pytest.mark.parametrize("klass", [ColumnLevelConstraint, ModelLevelConstraint])
-    def test_parse_constraint__invalid_column(self, klass):
+    @pytest.mark.parametrize(
+        "type, expectedType",
+        [
+            ("not_null", ColumnLevelConstraint),
+            ("check", CheckConstraint),
+            ("custom", CustomConstraint),
+            ("unique", ColumnLevelConstraint),
+            ("primary_key", PrimaryKeyConstraint),
+            ("foreign_key", ForeignKeyConstraint),
+        ],
+    )
+    def test_parse_constraint__valid_column(self, type, expectedType):
+        raw_constraint = {
+            "type": type,
+            "expression": "1 = 1",
+            "name": "my_constraint",
+        }
+        if type not in ["not_null", "unique"]:
+            raw_constraint["columns"] = ["id"]
+        constraint = parse_constraint(raw_constraint)
+        assert isinstance(constraint, expectedType)
+
+    def test_parse_constraint__invalid_constraint(self):
         raw_constraint = {"type": None}
         with pytest.raises(DbtValidationError, match="Could not parse constraint"):
-            constraints.parse_constraint(klass, raw_constraint)
-
-
-@pytest.fixture
-def constraint():
-    return ColumnLevelConstraint(type=ConstraintType.check)
+            parse_constraint(raw_constraint)
 
 
 class TestProcessConstraint:
-    @pytest.fixture
-    def success(self):
-        return "SUCCESS"
-
-    @pytest.fixture
-    def render_map(self, constraint, success):
-        return {constraint.type: lambda _: success}
-
-    def test_process_constraint__valid_constraint(self, constraint, render_map, success):
-        assert constraints.process_constraint(constraint, lambda _: True, render_map) == success
+    def test_process_constraint__valid_constraint(self):
+        constraint = CheckConstraint(type=ConstraintType.check, expression="1 = 1")
+        assert process_constraint(constraint) == constraint.render()
 
-    def test_process_constraint__invalid_constraint(self, constraint, render_map):
-        assert constraints.process_constraint(constraint, lambda x: False, render_map) is None
+    def test_process_constraint__invalid_constraint(self):
+        constraint = ColumnLevelConstraint(type=ConstraintType.unique)
+        assert process_constraint(constraint) is None
 
 
 class TestValidateConstraint:
@@ -92,215 +223,97 @@ class TestValidateConstraint:
 
     def test_validate_constraint__custom(self):
         constraint = ColumnLevelConstraint(type=ConstraintType.custom)
-        assert constraints.validate_constraint(constraint, lambda _: False) is True
+        assert validate_constraint(constraint) is True
 
     def test_validate_constraint__supported(self, pk_constraint):
-        assert constraints.validate_constraint(pk_constraint, lambda _: True) is True
+        assert validate_constraint(pk_constraint) is True
 
     @patch("dbt.adapters.databricks.constraints.warn_or_error")
-    def test_validate_constraint__unsupported(self, pk_constraint):
+    def test_validate_constraint__unsupported(self, _):
         with patch("dbt.adapters.databricks.constraints.ConstraintNotSupported") as mock_warn:
-            assert constraints.validate_constraint(pk_constraint, lambda _: False) is False
+            constraint = ColumnLevelConstraint(type=ConstraintType.unique)
+            assert validate_constraint(constraint) is False
             mock_warn.assert_called_with(
-                constraint=pk_constraint.type.value, adapter="DatabricksAdapter"
+                constraint=constraint.type.value, adapter="DatabricksAdapter"
             )
 
     @patch("dbt.adapters.databricks.constraints.warn_or_error")
-    def test_validate_constraint__unenforced(self, pk_constraint):
+    def test_validate_constraint__unenforced(self, _, pk_constraint):
         with patch("dbt.adapters.databricks.constraints.ConstraintNotEnforced") as mock_warn:
-            assert constraints.validate_constraint(pk_constraint, lambda _: True) is True
+            assert validate_constraint(pk_constraint) is True
             mock_warn.assert_called_with(
                 constraint=pk_constraint.type.value, adapter="DatabricksAdapter"
             )
 
 
-class TestRenderConstraint:
-    @pytest.fixture
-    def success(self):
-        return "CHECK (1 = 1)"
-
-    @pytest.fixture
-    def render_map(self, constraint, success):
-        return {constraint.type: lambda _: success}
+class TestParseConstraints:
+    def test_parse_column_constraints__empty(self):
+        assert (set(), []) == parse_column_constraints([])
 
-    def test_render_constraint__valid_constraint(self, constraint, render_map, success):
-        assert constraints.render_constraint(constraint, render_map) == success
+    def test_parse_column_constraints__not_nulls(self):
+        columns = [{"name": "id", "constraints": [{"type": "not_null"}]}]
+        assert ({"id"}, []) == parse_column_constraints(columns)
 
-    def test_render_constraint__invalid_constraint(self, render_map):
+    def test_parse_column_constraints__model_constraints(self):
+        columns = [{"name": "id", "constraints": [{"type": "primary_key"}]}]
         assert (
-            constraints.render_constraint(
-                ColumnLevelConstraint(type=ConstraintType.not_null), render_map
-            )
-            is None
-        )
-
-    def test_render_constraint__with_name(self, constraint, render_map, success):
-        constraint.name = "my_constraint"
-        assert (
-            constraints.render_constraint(constraint, render_map)
-            == f"CONSTRAINT {constraint.name} {success}"
-        )
+            set(),
+            [PrimaryKeyConstraint(type=ConstraintType.primary_key, columns=["id"])],
+        ) == parse_column_constraints(columns)
 
-    def test_render_constraint__excess_gets_trimmed(self, constraint):
-        constraint.name = "my_constraint"
-        render_map = {constraint.type: lambda _: "CHECK (1 = 1)   "}
+    def test_parse_column_constraints__both(self):
+        columns = [{"name": "id", "constraints": [{"type": "primary_key"}, {"type": "not_null"}]}]
         assert (
-            constraints.render_constraint(constraint, render_map)
-            == f"CONSTRAINT {constraint.name} CHECK (1 = 1)"
-        )
+            {"id"},
+            [PrimaryKeyConstraint(type=ConstraintType.primary_key, columns=["id"])],
+        ) == parse_column_constraints(columns)
 
+    def test_parse_model_constraints__empty(self):
+        assert (set(), []) == parse_model_constraints([])
 
-class TestSupportedFor:
-    @pytest.fixture
-    def warning(self):
-        return "Warning for {type}"
-
-    def test_supported_for__supported(self, constraint, warning):
-        assert constraints.supported_for(constraint, lambda _: True, warning) is True
-
-    def test_supported_for__not_supported_in_base(self, warning):
-        constraint = ColumnLevelConstraint(type=ConstraintType.unique)
-        assert constraints.supported_for(constraint, lambda _: True, warning) is False
-
-    def test_supported_for__not_supported_in_context(self, constraint, warning):
-        with patch("dbt.adapters.databricks.constraints.logger.warning") as mock_warn:
-            assert constraints.supported_for(constraint, lambda _: False, warning) is False
-            mock_warn.assert_called_with(warning.format(type=constraint.type))
-
-
-class TestRenderError:
-    @pytest.mark.parametrize(
-        "constraint, missing, expected",
-        [
-            (
-                ColumnLevelConstraint(type=ConstraintType.check),
-                [["expression"]],
-                "check constraint is missing required field(s): ('expression')",
-            ),
-            (
-                ColumnLevelConstraint(type=ConstraintType.foreign_key),
-                [["expression"], ["to", "to_columns"]],
-                (
-                    "foreign_key constraint is missing required field(s): "
-                    "('expression') or ('to', 'to_columns')"
-                ),
-            ),
-            (
-                ColumnLevelConstraint(type=ConstraintType.primary_key, name="my_pk"),
-                [["expression"]],
-                "primary_key constraint my_pk is missing required field(s): ('expression')",
-            ),
-        ],
-    )
-    def test_render_error__expected(self, constraint, missing, expected):
-        assert constraints.render_error(constraint, missing).msg == expected
-
-
-class TestRenderCustom:
-    def test_render_custom__valid(self):
-        constraint = ColumnLevelConstraint(type=ConstraintType.custom, expression="1 = 1")
-        assert constraints.render_custom(constraint) == "1 = 1"
-
-    def test_render_custom__invalid(self):
-        constraint = ColumnLevelConstraint(type=ConstraintType.custom)
-        with pytest.raises(DbtValidationError, match="custom constraint is missing required field"):
-            constraints.render_custom(constraint)
-
-
-class TestRenderPrimaryKeyForColumn:
-    def test_render_primary_key_for_column__valid(self):
-        constraint = ColumnLevelConstraint(type=ConstraintType.primary_key, expression="DEFERRABLE")
-        assert constraints.render_primary_key_for_column(constraint) == "PRIMARY KEY DEFERRABLE"
-
-    def test_render_primary_key_for_column__no_expression(self):
-        constraint = ColumnLevelConstraint(type=ConstraintType.primary_key)
-        assert constraints.render_primary_key_for_column(constraint) == "PRIMARY KEY"
-
-
-class TestRenderForeignKeyForColumn:
-    def test_render_foreign_key_for_column__valid_to(self):
-        constraint = ColumnLevelConstraint(
-            type=ConstraintType.foreign_key,
-            to="other_table",
-            to_columns=["other_id"],
-        )
-        assert (
-            constraints.render_foreign_key_for_column(constraint)
-            == "FOREIGN KEY REFERENCES other_table (other_id)"
-        )
-
-    def test_render_foreign_key_for_column__valid_expression(self):
-        constraint = ColumnLevelConstraint(
-            type=ConstraintType.foreign_key, expression="references other_table (other_id)"
-        )
-        assert (
-            constraints.render_foreign_key_for_column(constraint)
-            == "FOREIGN KEY references other_table (other_id)"
-        )
-
-    def test_render_foreign_key_for_column__invalid(self):
-        constraint = ColumnLevelConstraint(type=ConstraintType.foreign_key)
-        with pytest.raises(
-            DbtValidationError, match="foreign_key constraint is missing required field"
-        ):
-            constraints.render_foreign_key_for_column(constraint)
-
-
-class TestRenderPrimaryKeyForModel:
-    def test_render_primary_key_for_model__valid_columns(self):
-        constraint = ModelLevelConstraint(type=ConstraintType.primary_key, columns=["id", "ts"])
-        assert constraints.render_primary_key_for_model(constraint) == "PRIMARY KEY (id, ts)"
-
-    def test_render_primary_key_for_model__expression(self):
-        constraint = ModelLevelConstraint(
-            type=ConstraintType.primary_key, expression="(id TIMESERIES)"
-        )
-        assert constraints.render_primary_key_for_model(constraint) == "PRIMARY KEY (id TIMESERIES)"
-
-    def test_render_primary_key_for_model__invalid(self):
-        constraint = ModelLevelConstraint(type=ConstraintType.primary_key)
+    def test_parse_model_constraints__not_nulls_invalid(self):
+        constraints = [{"type": "not_null"}]
         with pytest.raises(
-            DbtValidationError, match="primary_key constraint is missing required field"
+            DbtValidationError, match="not_null constraint on model must have 'columns' defined"
         ):
-            constraints.render_primary_key_for_model(constraint)
+            parse_model_constraints(constraints)
 
+    def test_parse_model_constraints__not_nulls(self):
+        constraints = [{"type": "not_null", "columns": ["id"]}]
+        assert ({"id"}, []) == parse_model_constraints(constraints)
 
-class TestRenderForeignKeyForModel:
-    def test_render_foreign_key_for_model__valid_columns(self):
-        constraint = ModelLevelConstraint(
-            type=ConstraintType.foreign_key,
-            columns=["id"],
-            to="other_table",
-            to_columns=["other_id"],
-        )
+    def test_parse_model_constraints__model_constraints(self):
+        columns = [{"type": "primary_key", "columns": ["id"]}]
         assert (
-            constraints.render_foreign_key_for_model(constraint)
-            == "FOREIGN KEY (id) REFERENCES other_table (other_id)"
-        )
-
-    def test_render_foreign_key_for_model__expression(self):
-        constraint = ModelLevelConstraint(
-            type=ConstraintType.foreign_key, expression="references other_table (other_id)"
-        )
+            set(),
+            [PrimaryKeyConstraint(type=ConstraintType.primary_key, columns=["id"])],
+        ) == parse_model_constraints(columns)
+
+    def test_parse_model_constraints__both(self):
+        columns = [
+            {"type": "primary_key", "columns": ["id"]},
+            {"type": "not_null", "columns": ["id"]},
+        ]
         assert (
-            constraints.render_foreign_key_for_model(constraint)
-            == "FOREIGN KEY references other_table (other_id)"
-        )
-
-    def test_render_foreign_key_for_model__invalid(self):
-        constraint = ModelLevelConstraint(type=ConstraintType.foreign_key)
-        with pytest.raises(
-            DbtValidationError, match="foreign_key constraint is missing required field"
-        ):
-            constraints.render_foreign_key_for_model(constraint)
+            {"id"},
+            [PrimaryKeyConstraint(type=ConstraintType.primary_key, columns=["id"])],
+        ) == parse_model_constraints(columns)
 
+    def test_parse_constraints__empty(self):
+        assert (set(), []) == parse_constraints([], [])
 
-class TestRenderCheck:
-    def test_render_check__valid(self):
-        constraint = ModelLevelConstraint(type=ConstraintType.check, expression="id > 0")
-        assert constraints.render_check(constraint) == "CHECK (id > 0)"
+    def test_parse_constraints__not_nulls(self):
+        columns = [{"name": "id", "constraints": [{"type": "not_null"}]}]
+        constraints = [{"type": "not_null", "columns": ["id2"]}]
+        assert ({"id", "id2"}, []) == parse_constraints(columns, constraints)
 
-    def test_render_check__invalid(self):
-        constraint = ModelLevelConstraint(type=ConstraintType.check)
-        with pytest.raises(DbtValidationError, match="check constraint is missing required field"):
-            constraints.render_check(constraint)
+    def test_parse_constraints__constraints(self):
+        columns = [{"name": "id", "constraints": [{"type": "primary_key"}]}]
+        constraints = [{"type": "custom", "expression": "1 = 1"}]
+        assert (
+            set(),
+            [
+                PrimaryKeyConstraint(type=ConstraintType.primary_key, columns=["id"]),
+                CustomConstraint(type=ConstraintType.custom, expression="1 = 1"),
+            ],
+        ) == parse_constraints(columns, constraints)
diff --git a/tests/unit/test_relation.py b/tests/unit/test_relation.py
index ab15cecf..f18f639f 100644
--- a/tests/unit/test_relation.py
+++ b/tests/unit/test_relation.py
@@ -1,7 +1,12 @@
 import pytest
-from dbt_common.contracts.constraints import ConstraintType, ModelLevelConstraint
+from dbt_common.contracts.constraints import ConstraintType
 
 from dbt.adapters.databricks import relation
+from dbt.adapters.databricks.constraints import (
+    CheckConstraint,
+    CustomConstraint,
+    PrimaryKeyConstraint,
+)
 from dbt.adapters.databricks.relation import DatabricksQuotePolicy, DatabricksRelation
 
 
@@ -211,15 +216,15 @@ class TestConstraints:
 
     @pytest.fixture
     def custom_constraint(self):
-        return ModelLevelConstraint(type=ConstraintType.custom, expression="a > 1")
+        return CustomConstraint(type=ConstraintType.custom, expression="a > 1")
 
     @pytest.fixture
     def check_constraint(self):
-        return ModelLevelConstraint(type=ConstraintType.check, expression="a > 1")
+        return CheckConstraint(type=ConstraintType.check, expression="a > 1")
 
     @pytest.fixture
     def pk_constraint(self):
-        return ModelLevelConstraint(type=ConstraintType.primary_key, columns=["a"])
+        return PrimaryKeyConstraint(type=ConstraintType.primary_key, columns=["a"])
 
     def test_add_constraint__check_is_an_alter_constraint(self, relation, check_constraint):
         relation.add_constraint(check_constraint)
@@ -240,9 +245,7 @@ class TestConstraints:
         assert id(enriched) != id(relation)
 
     def test_enrich_relation__adds_constraints(self, relation, check_constraint, custom_constraint):
-        enriched = relation.enrich(
-            [{"type": "check", "expression": "a > 1"}, {"type": "custom", "expression": "a > 1"}]
-        )
+        enriched = relation.enrich([check_constraint, custom_constraint])
         assert enriched.alter_constraints == [check_constraint]
         assert enriched.create_constraints == [custom_constraint]