Fix: disable query validations for dbt models

Author: georgesittas

sqlmesh/core/context.py

@@ -81,7 +81,7 @@
 from sqlmesh.core.linter.rules import BUILTIN_RULES
 from sqlmesh.core.macros import ExecutableOrMacro, macro
 from sqlmesh.core.metric import Metric, rewrite
-from sqlmesh.core.model import Model, update_model_schemas
+from sqlmesh.core.model import Model, SqlModel, update_model_schemas
 from sqlmesh.core.config.model import ModelDefaultsConfig
 from sqlmesh.core.notification_target import (
     NotificationEvent,
@@ -681,10 +681,13 @@ def load(self, update_schemas: bool = True) -> GenericContext[C]:
                 cache_dir=self.cache_dir,
             )
 
-            models = self.models.values()
-            for model in models:
-                # The model definition can be validated correctly only after the schema is set.
-                model.validate_definition()
+            # The model definition can be validated correctly only after the schema is set.
+            for model in self.models.values():
+                if isinstance(model, SqlModel):
+                    # Validates only the model metadata; SQL validations are handled by the linter
+                    super(SqlModel, model).validate_definition()  # type: ignore
+                else:
+                    model.validate_definition()
 
         duplicates = set(self._models) & set(self._standalone_audits)
         if duplicates:

sqlmesh/core/linter/rules/builtin.py

@@ -25,6 +25,7 @@
 )
 from sqlmesh.core.linter.definition import RuleSet
 from sqlmesh.core.model import Model, SqlModel, ExternalModel
+from sqlmesh.utils.errors import ConfigError
 from sqlmesh.utils.lineage import extract_references_from_query, ExternalModelReference
 
 
@@ -274,4 +275,22 @@ def create_fix(self, model_name: str) -> t.Optional[Fix]:
         )
 
 
+class ValidateModelDefinition(Rule):
+    """
+    Checks whether a model satisfies certain properties, such as (but not limited to):
+
+    - If SQL-based, it contains at least one projection & projection names are unique
+    - Its kind is configured correctly (e.g., the VIEW kind is not supported for Python models)
+    - Other metadata properties are well-formed (e.g., incremental-by-time models require a time column)
+    """
+
+    def check_model(self, model: Model) -> t.Optional[RuleViolation]:
+        try:
+            model.validate_definition()
+        except ConfigError as ex:
+            return self.violation(str(ex))
+
+        return None
+
+
 BUILTIN_RULES = RuleSet(subclasses(__name__, Rule, (Rule,)))

sqlmesh/core/model/definition.py

@@ -1417,12 +1417,20 @@ def columns_to_types(self) -> t.Optional[t.Dict[str, exp.DataType]]:
 
             unknown = exp.DataType.build("unknown")
 
-            self._columns_to_types = {
+            columns_to_types = {}
+            for select in query.selects:
+                output_name = select.output_name
+
+                # If model validation is disabled, we cannot assume that projections
+                # will have inferrable output names or even that they will be unique
+                if not output_name or output_name in columns_to_types:
+                    return None
+
                 # copy data type because it is used in the engine to build CTAS and other queries
                 # this can change the parent which will mess up the diffing algo
-                select.output_name: (select.type or unknown).copy()
-                for select in query.selects
-            }
+                columns_to_types[output_name] = (select.type or unknown).copy()
+
+            self._columns_to_types = columns_to_types
 
         if "*" in self._columns_to_types:
             return None
@@ -1846,8 +1854,9 @@ def validate_definition(self) -> None:
         super().validate_definition()
 
         if self.kind and not self.kind.supports_python_models:
-            raise SQLMeshError(
-                f"Cannot create Python model '{self.name}' as the '{self.kind.name}' kind doesn't support Python models"
+            raise_config_error(
+                f"Cannot create Python model '{self.name}' as the '{self.kind.name}' kind doesn't support Python models",
+                self._path,
             )
 
     def render(

tests/core/test_context.py

@@ -1554,6 +1554,38 @@ def test_raw_code_handling(sushi_test_dbt_context: Context):
     )
 
 
+@pytest.mark.slow
+def test_dbt_models_are_not_validated(sushi_test_dbt_context: Context):
+    model = sushi_test_dbt_context.models['"memory"."sushi"."non_validated_model"']
+
+    assert model.render_query_or_raise().sql(comments=False) == 'SELECT 1 AS "c", 2 AS "c"'
+    assert sushi_test_dbt_context.fetchdf(
+        'SELECT * FROM "memory"."sushi"."non_validated_model"'
+    ).to_dict() == {"c": {0: 1}, "c_1": {0: 2}}
+
+    # Write a new incremental model file that should fail validation
+    models_dir = sushi_test_dbt_context.path / "models"
+    incremental_model_path = models_dir / "invalid_incremental.sql"
+    incremental_model_content = """{{
+  config(
+    materialized='incremental',
+    incremental_strategy='delete+insert',
+  )
+}}
+
+SELECT
+  1 AS c"""
+
+    incremental_model_path.write_text(incremental_model_content)
+
+    # Reload the context - this should raise a validation error for the incremental model
+    with pytest.raises(
+        ConfigError,
+        match="Unmanaged incremental models with insert / overwrite enabled must specify the partitioned_by field",
+    ):
+        Context(paths=sushi_test_dbt_context.path, config="test_config")
+
+
 def test_catalog_name_needs_to_be_quoted():
     config = Config(
         model_defaults=ModelDefaultsConfig(dialect="duckdb"),

tests/fixtures/dbt/sushi_test/models/non_validated_model.sql

@@ -0,0 +1,5 @@
+{{ config(materialized='table') }}
+
+SELECT
+  1 AS c,
+  2 AS c,