Detect breaking changes to column names and data types in state:modified check

.changes/unreleased/Features-20230323-133026.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Detect breaking changes to contracts in state:modified check
+time: 2023-03-23T13:30:26.593717-04:00
+custom:
+  Author: gshank
+  Issue: "6869"

.changes/unreleased/Under the Hood-20230217-105223.yaml

@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Treat contract config as a python object
+time: 2023-02-17T10:52:23.212474-05:00
+custom:
+  Author: gshank emmyoop
+  Issue: 6748 7184

core/dbt/contracts/graph/model_config.py

@@ -189,6 +189,11 @@ class Severity(str):
 register_pattern(Severity, insensitive_patterns("warn", "error"))
 
 
+@dataclass
+class ContractConfig(dbtClassMixin, Replaceable):
+    enforced: bool = False
+
+
 @dataclass
 class Hook(dbtClassMixin, Replaceable):
     sql: str
@@ -286,7 +291,7 @@ def same_contents(cls, unrendered: Dict[str, Any], other: Dict[str, Any]) -> boo
     # 'meta' moved here from node
     mergebehavior = {
         "append": ["pre-hook", "pre_hook", "post-hook", "post_hook", "tags"],
-        "update": ["quoting", "column_types", "meta", "docs"],
+        "update": ["quoting", "column_types", "meta", "docs", "contract"],
         "dict_key_append": ["grants"],
     }
 
@@ -451,7 +456,10 @@ class NodeConfig(NodeAndTestConfig):
         default_factory=Docs,
         metadata=MergeBehavior.Update.meta(),
     )
-    contract: bool = False
+    contract: ContractConfig = field(
+        default_factory=ContractConfig,
+        metadata=MergeBehavior.Update.meta(),
+    )
 
     # we validate that node_color has a suitable value to prevent dbt-docs from crashing
     def __post_init__(self):

core/dbt/contracts/graph/nodes.py

@@ -2,6 +2,7 @@
 import time
 from dataclasses import dataclass, field
 from enum import Enum
+import hashlib
 
 from mashumaro.types import SerializableType
 from typing import (
@@ -20,25 +21,25 @@
 from dbt.clients.system import write_file
 from dbt.contracts.files import FileHash
 from dbt.contracts.graph.unparsed import (
-    Quoting,
     Docs,
-    FreshnessThreshold,
+    ExposureType,
     ExternalTable,
+    FreshnessThreshold,
     HasYamlMetadata,
     MacroArgument,
-    UnparsedSourceDefinition,
-    UnparsedSourceTableDefinition,
-    UnparsedColumn,
-    TestDef,
-    Owner,
-    ExposureType,
     MaturityType,
     MetricFilter,
     MetricTime,
+    Owner,
+    Quoting,
+    TestDef,
+    UnparsedSourceDefinition,
+    UnparsedSourceTableDefinition,
+    UnparsedColumn,
 )
 from dbt.contracts.util import Replaceable, AdditionalPropertiesMixin
 from dbt.events.functions import warn_or_error
-from dbt.exceptions import ParsingError, InvalidAccessTypeError
+from dbt.exceptions import ParsingError, InvalidAccessTypeError, ModelContractError
 from dbt.events.types import (
     SeedIncreased,
     SeedExceedsLimitSamePath,
@@ -50,7 +51,6 @@
 from dbt.flags import get_flags
 from dbt.node_types import ModelLanguage, NodeType, AccessType
 
-
 from .model_config import (
     NodeConfig,
     SeedConfig,
@@ -184,6 +184,12 @@ class ColumnInfo(AdditionalPropertiesMixin, ExtensibleDbtClassMixin, Replaceable
     _extra: Dict[str, Any] = field(default_factory=dict)
 
 
+@dataclass
+class Contract(dbtClassMixin, Replaceable):
+    enforced: bool = False
+    checksum: Optional[str] = None
+
+
 # Metrics, exposures,
 @dataclass
 class HasRelationMetadata(dbtClassMixin, Replaceable):
@@ -382,6 +388,13 @@ def same_config(self, old) -> bool:
             old.unrendered_config,
         )
 
+    def build_contract_checksum(self):
+        pass
+
+    def same_contract(self, old) -> bool:
+        # This would only apply to seeds
+        return True
+
     def patch(self, patch: "ParsedNodePatch"):
         """Given a ParsedNodePatch, add the new information to the node."""
         # explicitly pick out the parts to update so we don't inadvertently
@@ -423,6 +436,7 @@ def same_contents(self, old) -> bool:
             and self.same_persisted_description(old)
             and self.same_fqn(old)
             and self.same_database_representation(old)
+            and self.same_contract(old)
             and True
         )
 
@@ -451,7 +465,7 @@ class CompiledNode(ParsedNode):
     extra_ctes_injected: bool = False
     extra_ctes: List[InjectedCTE] = field(default_factory=list)
     _pre_injected_sql: Optional[str] = None
-    contract: bool = False
+    contract: Contract = field(default_factory=Contract)
 
     @property
     def empty(self):
@@ -492,6 +506,46 @@ def depends_on_nodes(self):
     def depends_on_macros(self):
         return self.depends_on.macros
 
+    def build_contract_checksum(self):
+        # We don't need to construct the checksum if the model does not
+        # have contract enforced, because it won't be used.
+        # This needs to be executed after contract config is set
+        if self.contract.enforced is True:
+            contract_state = ""
+            # We need to sort the columns so that order doesn't matter
+            # columns is a str: ColumnInfo dictionary
+            sorted_columns = sorted(self.columns.values(), key=lambda col: col.name)
+            for column in sorted_columns:
+                contract_state += f"|{column.name}"
+                contract_state += str(column.data_type)
+            data = contract_state.encode("utf-8")
+            self.contract.checksum = hashlib.new("sha256", data).hexdigest()
+
+    def same_contract(self, old) -> bool:
+        if old.contract.enforced is False and self.contract.enforced is False:
+            # Not a change
+            return True
+        if old.contract.enforced is False and self.contract.enforced is True:
+            # A change, but not a breaking change
+            return False
+
+        breaking_change_reasons = []
+        if old.contract.enforced is True and self.contract.enforced is False:
+            # Breaking change: throw an error
+            # Note: we don't have contract.checksum for current node, so build
+            self.build_contract_checksum()
+            breaking_change_reasons.append("contract has been disabled")
+
+        if self.contract.checksum != old.contract.checksum:
+            # Breaking change, throw error
+            breaking_change_reasons.append("column definitions have changed")
+
+        if breaking_change_reasons:
+            raise (ModelContractError(reasons=" and ".join(breaking_change_reasons), node=self))
+        else:
+            # no breaking changes
+            return True
+
 
 # ====================================
 # CompiledNode subclasses
@@ -914,7 +968,7 @@ def same_contents(self, old: Optional["SourceDefinition"]) -> bool:
         if old is None:
             return True
 
-        # config changes are changes (because the only config is "enabled", and
+        # config changes are changes (because the only config is "enforced", and
         # enabling a source is a change!)
         # changing the database/schema/identifier is a change
         # messing around with external stuff is a change (uh, right?)

core/dbt/exceptions.py

@@ -209,6 +209,25 @@ def _fix_dupe_msg(self, path_1: str, path_2: str, name: str, type_name: str) ->
             )
 
 
+class ModelContractError(DbtRuntimeError):
+    CODE = 10016
+    MESSAGE = "Contract Error"
+
+    def __init__(self, reasons, node=None):
+        self.reasons = reasons
+        super().__init__(self.message(), node)
+
+    @property
+    def type(self):
+        return "Contract"
+
+    def message(self):
+        return (
+            f"There is a breaking change in the model contract because {self.reasons}; "
+            "you may need to create a new version. See: https://docs.getdbt.com/docs/collaborate/publish/model-versions"
+        )
+
+
 class RecursionError(DbtRuntimeError):
     pass
 

core/dbt/graph/selector_methods.py

@@ -542,6 +542,7 @@ def search(self, included_nodes: Set[UniqueId], selector: str) -> Iterator[Uniqu
             ),
             "modified.relation": self.check_modified_factory("same_database_representation"),
             "modified.macros": self.check_modified_macros,
+            "modified.contract": self.check_modified_factory("same_contract"),
         }
         if selector in state_checks:
             checker = state_checks[selector]

core/dbt/include/global_project/macros/materializations/models/table/create_table_as.sql

@@ -25,7 +25,8 @@
 
   create {% if temporary: -%}temporary{%- endif %} table
     {{ relation.include(database=(not temporary), schema=(not temporary)) }}
-  {% if config.get('contract', False) %}
+  {% set contract_config = config.get('contract') %}
+  {% if contract_config.enforced %}
     {{ get_assert_columns_equivalent(sql) }}
     {{ get_columns_spec_ddl() }}
     {%- set sql = get_select_subquery(sql) %}

core/dbt/include/global_project/macros/materializations/models/view/create_view_as.sql

@@ -17,7 +17,8 @@
 
   {{ sql_header if sql_header is not none }}
   create view {{ relation }}
-    {%- if config.get('contract', False) %}
+    {% set contract_config = config.get('contract') %}
+    {% if contract_config.enforced %}
       {{ get_assert_columns_equivalent(sql) }}
     {%- endif %}
   as (

core/dbt/parser/base.py

@@ -16,8 +16,8 @@
 from dbt.config import Project, RuntimeConfig
 from dbt.context.context_config import ContextConfig
 from dbt.contracts.graph.manifest import Manifest
-from dbt.contracts.graph.nodes import ManifestNode, BaseNode
-from dbt.contracts.graph.unparsed import UnparsedNode, Docs
+from dbt.contracts.graph.nodes import Contract, BaseNode, ManifestNode
+from dbt.contracts.graph.unparsed import Docs, UnparsedNode
 from dbt.exceptions import DbtInternalError, ConfigUpdateError, DictParseError
 from dbt import hooks
 from dbt.node_types import NodeType, ModelLanguage
@@ -297,7 +297,7 @@ def update_parsed_node_config(
         # If we have docs in the config, merge with the node level, for backwards
         # compatibility with earlier node-only config.
         if "docs" in config_dict and config_dict["docs"]:
-            # we set show at the value of the config if it is set, otherwize, inherit the value
+            # we set show at the value of the config if it is set, otherwise, inherit the value
             docs_show = (
                 config_dict["docs"]["show"]
                 if "show" in config_dict["docs"]
@@ -310,10 +310,9 @@ def update_parsed_node_config(
             else:
                 parsed_node.docs = Docs(show=docs_show)
 
-        # If we have "contract" in the config, copy to node level, for backwards
-        # compatibility with earlier node-only config.
-        if config_dict.get("contract", False):
-            parsed_node.contract = True
+        # If we have contract in the config, copy to node level
+        if "contract" in config_dict and config_dict["contract"]:
+            parsed_node.contract = Contract(enforced=config_dict["contract"]["enforced"])
 
         # unrendered_config is used to compare the original database/schema/alias
         # values and to handle 'same_config' and 'same_contents' calls

core/dbt/parser/schemas.py

@@ -797,6 +797,7 @@ def get_unparsed_target(self) -> Iterable[NonSourceTarget]:
                     self.normalize_meta_attribute(data, path)
                     self.normalize_docs_attribute(data, path)
                     self.normalize_group_attribute(data, path)
+                    self.normalize_contract_attribute(data, path)
                 node = self._target_type().from_dict(data)
             except (ValidationError, JSONValidationError) as exc:
                 raise YamlParseDictError(path, self.key, data, exc)
@@ -828,6 +829,9 @@ def normalize_docs_attribute(self, data, path):
     def normalize_group_attribute(self, data, path):
         return self.normalize_attribute(data, path, "group")
 
+    def normalize_contract_attribute(self, data, path):
+        return self.normalize_attribute(data, path, "contract")
+
     def patch_node_config(self, node, patch):
         # Get the ContextConfig that's used in calculating the config
         # This must match the model resource_type that's being patched
@@ -937,10 +941,12 @@ def parse_patch(self, block: TargetBlock[NodeTarget], refs: ParserRef) -> None:
 
             node.patch(patch)
             self.validate_constraints(node)
+            node.build_contract_checksum()
 
     def validate_constraints(self, patched_node):
         error_messages = []
-        if patched_node.resource_type == "model" and patched_node.config.contract is True:
+        contract_config = patched_node.config.get("contract")
+        if patched_node.resource_type == "model" and contract_config.enforced is True:
             validators = [
                 self.constraints_schema_validator(patched_node),
                 self.constraints_materialization_validator(patched_node),

plugins/postgres/dbt/include/postgres/macros/adapters.sql

@@ -9,7 +9,8 @@
   {%- elif unlogged -%}
     unlogged
   {%- endif %} table {{ relation }}
-  {% if config.get('contract', False) %}
+  {% set contract_config = config.get('contract') %}
+  {% if contract_config.enforced %}
     {{ get_assert_columns_equivalent(sql) }}
     {{ get_columns_spec_ddl() }} ;
     insert into {{ relation }} {{ get_column_names() }}