dbt Constraints / model contracts

.changes/unreleased/Features-20221118-141120.yaml

@@ -0,0 +1,8 @@
+kind: Features
+body: Data type constraints are now native to SQL table materializations. Enforce
+  columns are specific data types and not null depending on database functionality.
+time: 2022-11-18T14:11:20.868062-08:00
+custom:
+  Author: sungchun12
+  Issue: "6079"
+  PR: "6271"

core/dbt/contracts/graph/model_config.py

@@ -446,6 +446,7 @@ class NodeConfig(NodeAndTestConfig):
         default_factory=Docs,
         metadata=MergeBehavior.Update.meta(),
     )
+    constraints_enabled: Optional[bool] = False
 
     # 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

@@ -61,6 +61,7 @@
     SnapshotConfig,
 )
 
+
 # =====================================================================
 # This contains the classes for all of the nodes and node-like objects
 # in the manifest. In the "nodes" dictionary of the manifest we find
@@ -146,6 +147,8 @@ class ColumnInfo(AdditionalPropertiesMixin, ExtensibleDbtClassMixin, Replaceable
     description: str = ""
     meta: Dict[str, Any] = field(default_factory=dict)
     data_type: Optional[str] = None
+    constraints: Optional[List[str]] = None
+    constraints_check: Optional[str] = None
     quote: Optional[bool] = None
     tags: List[str] = field(default_factory=list)
     _extra: Dict[str, Any] = field(default_factory=dict)
@@ -400,6 +403,7 @@ class CompiledNode(ParsedNode):
     extra_ctes_injected: bool = False
     extra_ctes: List[InjectedCTE] = field(default_factory=list)
     _pre_injected_sql: Optional[str] = None
+    constraints_enabled: bool = False
 
     @property
     def empty(self):

core/dbt/contracts/graph/unparsed.py

@@ -93,6 +93,8 @@ class HasDocs(AdditionalPropertiesMixin, ExtensibleDbtClassMixin, Replaceable):
     description: str = ""
     meta: Dict[str, Any] = field(default_factory=dict)
     data_type: Optional[str] = None
+    constraints: Optional[List[str]] = None
+    constraints_check: Optional[str] = None
     docs: Docs = field(default_factory=Docs)
     _extra: Dict[str, Any] = field(default_factory=dict)
 

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

@@ -0,0 +1,51 @@
+{%- macro get_columns_spec_ddl() -%}
+  {{ adapter.dispatch('get_columns_spec_ddl', 'dbt')() }}
+{%- endmacro -%}
+
+{% macro default__get_columns_spec_ddl() -%}
+  {{ return(columns_spec_ddl()) }}
+{%- endmacro %}
+
+{% macro columns_spec_ddl() %}
+  {# loop through user_provided_columns to create DDL with data types and constraints #}
+    {%- set user_provided_columns = model['columns'] -%}
+    (
+    {% for i in user_provided_columns %}
+      {% set col = user_provided_columns[i] %}
+      {% set constraints = col['constraints'] %}
+      {% set constraints_check = col['constraints_check'] %}
+      {{ col['name'] }} {{ col['data_type'] }} {% for x in constraints %} {{ x or "" }} {% endfor %} {% if constraints_check -%} check {{ constraints_check or "" }} {%- endif %} {{ "," if not loop.last }}
+    {% endfor %}
+  )
+{% endmacro %}
+
+{%- macro get_assert_columns_equivalent(sql) -%}
+  {{ adapter.dispatch('get_assert_columns_equivalent', 'dbt')(sql) }}
+{%- endmacro -%}
+
+{% macro default__get_assert_columns_equivalent(sql) -%}
+  {{ return(assert_columns_equivalent(sql)) }}
+{%- endmacro %}
+
+{% macro assert_columns_equivalent(sql) %}
+  {#- loop through user_provided_columns to get column names -#}
+    {%- set user_provided_columns = model['columns'] -%}
+    {%- set column_names_config_only = [] -%}
+    {%- for i in user_provided_columns -%}
+      {%- set col = user_provided_columns[i] -%}
+      {%- set col_name = col['name'] -%}
+      {%- set column_names_config_only = column_names_config_only.append(col_name) -%}
+    {%- endfor -%}
+    {%- set sql_file_provided_columns = get_columns_in_query(sql) -%}
+
+    {#- uppercase both schema and sql file columns -#}
+    {%- set column_names_config_upper= column_names_config_only|map('upper')|join(',')  -%}
+    {%- set column_names_config_formatted = column_names_config_upper.split(',')  -%}
+    {%- set sql_file_provided_columns_upper = sql_file_provided_columns|map('upper')|join(',') -%}
+    {%- set sql_file_provided_columns_formatted = sql_file_provided_columns_upper.split(',') -%}
+
+    {%- if column_names_config_formatted != sql_file_provided_columns_formatted -%}
+      {%- do exceptions.raise_compiler_error('Please ensure the name, order, and number of columns in your `yml` file match the columns in your SQL file.\nSchema File Columns: ' ~ column_names_config_formatted ~ '\nSQL File Columns: ' ~ sql_file_provided_columns_formatted ~ ' ' ) %}
+    {%- endif -%}
+
+{% endmacro %}

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

@@ -25,6 +25,10 @@
 
   create {% if temporary: -%}temporary{%- endif %} table
     {{ relation.include(database=(not temporary), schema=(not temporary)) }}
+    {% if config.get('constraints_enabled', False) %}
+      {{ get_assert_columns_equivalent(sql) }}
+      {{ get_columns_spec_ddl() }}
+    {% endif %}
   as (
     {{ sql }}
   );

core/dbt/parser/base.py

@@ -18,7 +18,7 @@
 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.exceptions import DbtInternalError, ConfigUpdateError, DictParseError
+from dbt.exceptions import DbtInternalError, ConfigUpdateError, DictParseError, ParsingError
 from dbt import hooks
 from dbt.node_types import NodeType, ModelLanguage
 from dbt.parser.search import FileBlock
@@ -306,6 +306,19 @@ def update_parsed_node_config(
             else:
                 parsed_node.docs = Docs(show=docs_show)
 
+        # If we have constraints_enabled in the config, copy to node level, for backwards
+        # compatibility with earlier node-only config.
+        if config_dict.get("constraints_enabled", False):
+            parsed_node.constraints_enabled = True
+
+            parser_name = type(self).__name__
+            if parser_name == "ModelParser":
+                original_file_path = parsed_node.original_file_path
+                error_message = "\n    `constraints_enabled=true` can only be configured within `schema.yml` files\n      NOT within a model file(ex: .sql, .py) or `dbt_project.yml`."
+                raise ParsingError(
+                    f"Original File Path: ({original_file_path})\nConstraints must be defined in a `yml` schema configuration file like `schema.yml`.\nOnly the SQL table materialization is supported for constraints. \n`data_type` values must be defined for all columns and NOT be null or blank.{error_message}"
+                )
+
         # unrendered_config is used to compare the original database/schema/alias
         # values and to handle 'same_config' and 'same_contents' calls
         parsed_node.unrendered_config = config.build_config_dict(

core/dbt/parser/schemas.py

@@ -119,6 +119,8 @@ def add(
         column: Union[HasDocs, UnparsedColumn],
         description: str,
         data_type: Optional[str],
+        constraints: Optional[List[str]],
+        constraints_check: Optional[str],
         meta: Dict[str, Any],
     ):
         tags: List[str] = []
@@ -132,6 +134,8 @@ def add(
             name=column.name,
             description=description,
             data_type=data_type,
+            constraints=constraints,
+            constraints_check=constraints_check,
             meta=meta,
             tags=tags,
             quote=quote,
@@ -144,8 +148,10 @@ def from_target(cls, target: Union[HasColumnDocs, HasColumnTests]) -> "ParserRef
         for column in target.columns:
             description = column.description
             data_type = column.data_type
+            constraints = column.constraints
+            constraints_check = column.constraints_check
             meta = column.meta
-            refs.add(column, description, data_type, meta)
+            refs.add(column, description, data_type, constraints, constraints_check, meta)
         return refs
 
 
@@ -914,6 +920,75 @@ def parse_patch(self, block: TargetBlock[NodeTarget], refs: ParserRef) -> None:
                 self.patch_node_config(node, patch)
 
             node.patch(patch)
+            self.validate_constraints(node)
+
+    def validate_constraints(self, patched_node):
+        error_messages = []
+        if (
+            patched_node.resource_type == "model"
+            and patched_node.config.constraints_enabled is True
+        ):
+            validators = [
+                self.constraints_schema_validator(patched_node),
+                self.constraints_materialization_validator(patched_node),
+                self.constraints_language_validator(patched_node),
+                self.constraints_data_type_validator(patched_node),
+            ]
+            error_messages = [validator for validator in validators if validator != "None"]
+
+        if error_messages:
+            original_file_path = patched_node.original_file_path
+            raise ParsingError(
+                f"Original File Path: ({original_file_path})\nConstraints must be defined in a `yml` schema configuration file like `schema.yml`.\nOnly the SQL table materialization is supported for constraints. \n`data_type` values must be defined for all columns and NOT be null or blank.{self.convert_errors_to_string(error_messages)}"
+            )
+
+    def convert_errors_to_string(self, error_messages: List[str]):
+        n = len(error_messages)
+        if not n:
+            return ""
+        if n == 1:
+            return error_messages[0]
+        error_messages_string = "".join(error_messages[:-1]) + f"{error_messages[-1]}"
+        return error_messages_string
+
+    def constraints_schema_validator(self, patched_node):
+        schema_error = False
+        if patched_node.columns == {}:
+            schema_error = True
+        schema_error_msg = "\n    Schema Error: `yml` configuration does NOT exist"
+        schema_error_msg_payload = f"{schema_error_msg if schema_error else None}"
+        return schema_error_msg_payload
+
+    def constraints_materialization_validator(self, patched_node):
+        materialization_error = {}
+        if patched_node.config.materialized != "table":
+            materialization_error = {"materialization": patched_node.config.materialized}
+        materialization_error_msg = f"\n    Materialization Error: {materialization_error}"
+        materialization_error_msg_payload = (
+            f"{materialization_error_msg if materialization_error else None}"
+        )
+        return materialization_error_msg_payload
+
+    def constraints_language_validator(self, patched_node):
+        language_error = {}
+        language = str(patched_node.language)
+        if language != "sql":
+            language_error = {"language": language}
+        language_error_msg = f"\n    Language Error: {language_error}"
+        language_error_msg_payload = f"{language_error_msg if language_error else None}"
+        return language_error_msg_payload
+
+    def constraints_data_type_validator(self, patched_node):
+        data_type_errors = set()
+        for column, column_info in patched_node.columns.items():
+            if column_info.data_type is None:
+                data_type_error = {column}
+                data_type_errors.update(data_type_error)
+        data_type_errors_msg = (
+            f"\n    Columns with `data_type` Blank/Null Errors: {data_type_errors}"
+        )
+        data_type_errors_msg_payload = f"{data_type_errors_msg if data_type_errors else None}"
+        return data_type_errors_msg_payload
 
 
 class TestablePatchParser(NodePatchParser[UnparsedNodeUpdate]):

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

@@ -9,7 +9,14 @@
   {%- elif unlogged -%}
     unlogged
   {%- endif %} table {{ relation }}
-  as (
+  {% if config.get('constraints_enabled', False) %}
+    {{ get_assert_columns_equivalent(sql) }}
+    {{ get_columns_spec_ddl() }} ;
+    insert into {{ relation }} {{ get_column_names() }}
+    {% else %}
+      as
+  {% endif %}
+    (
     {{ sql }}
   );
 {%- endmacro %}

plugins/postgres/dbt/include/postgres/macros/utils/columns_spec_ddl.sql

@@ -0,0 +1,23 @@
+{% macro postgres__get_columns_spec_ddl() %}
+  {# loop through user_provided_columns to create DDL with data types and constraints #}
+    {%- set user_provided_columns = model['columns'] -%}
+    (
+    {% for i in user_provided_columns %}
+      {% set col = user_provided_columns[i] %}
+      {% set constraints = col['constraints'] %}
+      {% set constraints_check = col['constraints_check'] %}
+      {{ col['name'] }} {{ col['data_type'] }} {% for x in constraints %} {{ x or "" }} {% endfor %} {% if constraints_check -%} check {{ constraints_check or "" }} {%- endif %} {{ "," if not loop.last }}
+    {% endfor %}
+  )
+{% endmacro %}
+
+{% macro get_column_names() %}
+  {# loop through user_provided_columns to get column names #}
+    {%- set user_provided_columns = model['columns'] -%}
+    (
+    {% for i in user_provided_columns %}
+      {% set col = user_provided_columns[i] %}
+      {{ col['name'] }} {{ "," if not loop.last }}
+    {% endfor %}
+  )
+{% endmacro %}