spike: inferred contracts

core/dbt/context/providers.py

@@ -1382,7 +1382,8 @@ def sql(self) -> Optional[str]:
         if self.model.language == ModelLanguage.sql:  # type: ignore[union-attr]
             # If the model is deferred and the adapter doesn't support zero-copy cloning, then select * from the prod
             # relation
-            if getattr(self.model, "defer_relation", None):
+            # TODO: avoid routing on args.which if possible
+            if getattr(self.model, "defer_relation", None) and self.config.args.which == "clone":
                 # TODO https://github.com/dbt-labs/dbt-core/issues/7976
                 return f"select * from {self.model.defer_relation.relation_name or str(self.defer_relation)}"  # type: ignore[union-attr]
             elif getattr(self.model, "extra_ctes_injected", None):

core/dbt/contracts/graph/manifest.py

@@ -1253,6 +1253,11 @@ def merge_from_artifact(
             ):
                 merged.add(unique_id)
                 self.nodes[unique_id] = node.replace(deferred=True)
+            elif current and (node.resource_type in refables and not node.is_ephemeral):
+                defer_relation = DeferRelation(
+                    node.database, node.schema, node.alias, node.relation_name
+                )
+                self.nodes[unique_id] = current.replace(defer_relation=defer_relation)
 
         # Rebuild the flat_graph, which powers the 'graph' context variable,
         # now that we've deferred some nodes

core/dbt/contracts/graph/model_config.py

@@ -204,6 +204,7 @@ def default(cls) -> "OnConfigurationChangeOption":
 @dataclass
 class ContractConfig(dbtClassMixin, Replaceable):
     enforced: bool = False
+    inferred: bool = False
 
 
 @dataclass

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

@@ -8,8 +8,22 @@
 
 {% macro table_columns_and_constraints() %}
   {# loop through user_provided_columns to create DDL with data types and constraints #}
+    {%- if defer_relation and config.get('contract').inferred -%}
+      {% set prod_columns = get_column_schema_from_query(get_empty_subquery_sql("select * from " ~ defer_relation)) %}
+      {% set raw_columns = {} %}
+      {% for column in prod_columns -%}
+        {# TODO: reconsider this mutating operation, could be done with an additional warehouse call #}
+        {%- if column.name not in model['columns'] -%}
+          {% do model['columns'].update(
+            {column.name: {'name': column.name, 'data_type': column.data_type, 'description': '', 'constraints': [], 'tags': [], 'meta': {}}}
+            ) %}
+          {%- endif -%}
+      {% endfor -%}
+    {%- endif -%}
+
     {%- set raw_column_constraints = adapter.render_raw_columns_constraints(raw_columns=model['columns']) -%}
     {%- set raw_model_constraints = adapter.render_raw_model_constraints(raw_constraints=model['constraints']) -%}
+
     (
     {% for c in raw_column_constraints -%}
       {{ c }}{{ "," if not loop.last or raw_model_constraints }}
@@ -36,15 +50,30 @@
 
   {#-- First ensure the user has defined 'columns' in yaml specification --#}
   {%- set user_defined_columns = model['columns'] -%}
-  {%- if not user_defined_columns -%}
+
+  {%- if not user_defined_columns and not config.get('contract').inferred -%}
       {{ exceptions.raise_contract_error([], []) }}
   {%- endif -%}
 
   {#-- Obtain the column schema provided by sql file. #}
   {%- set sql_file_provided_columns = get_column_schema_from_query(sql, config.get('sql_header', none)) -%}
   {#--Obtain the column schema provided by the schema file by generating an 'empty schema' query from the model's columns. #}
   {%- set schema_file_provided_columns = get_column_schema_from_query(get_empty_schema_sql(user_defined_columns)) -%}
+  {%- if defer_relation and config.get('contract').inferred -%}
+      {% set prod_columns = get_column_schema_from_query(get_empty_subquery_sql("select * from " ~ defer_relation)) %}
+
+      {% set provided_column_dict = {} %}
+      {%- for provided_column in  schema_file_provided_columns-%}
+        {%- do provided_column_dict.update({provided_column.name: provided_column}) %}
+      {%- endfor -%}
 
+      {#-- Add any inferred columns not provided to full list of provided columns #}
+      {%- for prod_colum in prod_columns -%}
+        {%- if prod_colum.name not in provided_column_dict -%}
+          {% do schema_file_provided_columns.append(prod_colum) -%}
+        {%- endif -%}
+      {%- endfor -%}
+  {%- endif -%}
   {#-- create dictionaries with name and formatted data type and strings for exception #}
   {%- set sql_columns = format_columns(sql_file_provided_columns) -%}
   {%- set yaml_columns = format_columns(schema_file_provided_columns)  -%}

core/dbt/parser/schemas.py

@@ -849,7 +849,7 @@ def _validate_constraint_prerequisites(self, model_node: ModelNode):
             )
 
         errors = []
-        if not model_node.columns:
+        if not model_node.columns and not model_node.config.contract.inferred:
             errors.append(
                 "Constraints must be defined in a `yml` schema configuration file like `schema.yml`."
             )

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

@@ -11,6 +11,9 @@
   {%- endif %} table {{ relation }}
   {% set contract_config = config.get('contract') %}
   {% if contract_config.enforced %}
+    {%- if contract_config.inferred and not defer_relation -%}
+      {{ exceptions.raise_compiler_error("Contract cannot be inferred without a --defer relation")}}
+    {%- endif -%}
     {{ get_assert_columns_equivalent(sql) }}
     {{ get_table_columns_and_constraints() }} ;
     insert into {{ relation }} (