Address PR comments

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, SqlModel, update_model_schemas
+from sqlmesh.core.model import Model, update_model_schemas
 from sqlmesh.core.config.model import ModelDefaultsConfig
 from sqlmesh.core.notification_target import (
     NotificationEvent,
@@ -681,13 +681,10 @@ def load(self, update_schemas: bool = True) -> GenericContext[C]:
                 cache_dir=self.cache_dir,
             )
 
-            # 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()
+            models = self.models.values()
+            for model in models:
+                # The model definition can be validated correctly only after the schema is set.
+                model.validate_definition()
 
         duplicates = set(self._models) & set(self._standalone_audits)
         if duplicates:
@@ -3134,7 +3131,9 @@ def lint_models(
         found_error = False
 
         model_list = (
-            list(self.get_model(model) for model in models) if models else self.models.values()
+            list(self.get_model(model, raise_if_missing=True) for model in models)
+            if models
+            else self.models.values()
         )
         all_violations = []
         for model in model_list:

sqlmesh/core/linter/rules/builtin.py

@@ -25,7 +25,6 @@
 )
 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
 
 
@@ -275,20 +274,31 @@ 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)
-    """
+class NoAmbiguousProjections(Rule):
+    """All projections in a model must have unique, inferrable names or explicit aliases."""
 
     def check_model(self, model: Model) -> t.Optional[RuleViolation]:
-        try:
-            model.validate_definition()
-        except ConfigError as ex:
-            return self.violation(str(ex))
+        query = model.render_query()
+        if query is None:
+            return None
+
+        name_counts: t.Dict[str, int] = {}
+        projection_list = query.selects
+        for expression in projection_list:
+            alias = expression.output_name
+            if alias == "*":
+                continue
+
+            if not alias:
+                return self.violation(
+                    f"Outer projection '{expression.sql(dialect=model.dialect)}' must have inferrable names or explicit aliases."
+                )
+
+            name_counts[alias] = name_counts.get(alias, 0) + 1
+
+        for name, count in name_counts.items():
+            if count > 1:
+                return self.violation(f"Found duplicate outer select name '{name}'")
 
         return None
 

sqlmesh/core/model/definition.py

@@ -1481,22 +1481,6 @@ def validate_definition(self) -> None:
         if not projection_list:
             raise_config_error("Query missing select statements", self._path)
 
-        name_counts: t.Dict[str, int] = {}
-        for expression in projection_list:
-            alias = expression.output_name
-            if alias == "*":
-                continue
-            if not alias:
-                raise_config_error(
-                    f"Outer projection '{expression.sql(dialect=self.dialect)}' must have inferrable names or explicit aliases.",
-                    self._path,
-                )
-            name_counts[alias] = name_counts.get(alias, 0) + 1
-
-        for name, count in name_counts.items():
-            if count > 1:
-                raise_config_error(f"Found duplicate outer select name '{name}'", self._path)
-
         if self.depends_on_self and not self.annotated:
             raise_config_error(
                 "Self-referencing models require inferrable column types. There are three options available to mitigate this issue: add explicit types to all projections in the outermost SELECT statement, leverage external models (https://sqlmesh.readthedocs.io/en/stable/concepts/models/external_models/), or use the `columns` model attribute (https://sqlmesh.readthedocs.io/en/stable/concepts/models/overview/#columns).",

tests/core/test_model.py

@@ -31,12 +31,12 @@
     ModelDefaultsConfig,
     LinterConfig,
 )
+from sqlmesh.core import constants as c
 from sqlmesh.core.context import Context, ExecutionContext
 from sqlmesh.core.dialect import parse
 from sqlmesh.core.engine_adapter.base import MERGE_SOURCE_ALIAS, MERGE_TARGET_ALIAS
 from sqlmesh.core.engine_adapter.duckdb import DuckDBEngineAdapter
 from sqlmesh.core.macros import MacroEvaluator, macro
-from sqlmesh.core import constants as c
 from sqlmesh.core.model import (
     CustomKind,
     PythonModel,
@@ -198,29 +198,64 @@ def test_model_multiple_select_statements():
         load_sql_based_model(expressions)
 
 
-@pytest.mark.parametrize(
-    "query, error",
-    [
-        ("y::int, x::int AS y", "duplicate"),
-        ("* FROM db.table", "require inferrable column types"),
-    ],
-)
-def test_model_validation(query, error):
+def test_model_validation(tmp_path):
     expressions = d.parse(
         f"""
         MODEL (
             name db.table,
             kind FULL,
         );
 
-        SELECT {query}
+        SELECT
+          y::int,
+          x::int AS y
+        FROM db.ext
+        """
+    )
+
+    ctx = Context(
+        config=Config(linter=LinterConfig(enabled=True, rules=["noambiguousprojections"])),
+        paths=tmp_path,
+    )
+    ctx.upsert_model(load_sql_based_model(expressions, default_catalog="memory"))
+
+    errors = ctx.lint_models(["db.table"], raise_on_error=False)
+    assert errors, "Expected NoAmbiguousProjections violation"
+    assert errors[0].violation_msg == "Found duplicate outer select name 'y'"
+
+    expressions = d.parse(
+        """
+        MODEL (
+            name db.table,
+            kind FULL,
+        );
+
+        SELECT a, a UNION SELECT c, c
+        """
+    )
+
+    ctx.upsert_model(load_sql_based_model(expressions, default_catalog="memory"))
+
+    errors = ctx.lint_models(["db.table"], raise_on_error=False)
+    assert errors, "Expected NoAmbiguousProjections violation"
+    assert errors[0].violation_msg == "Found duplicate outer select name 'a'"
+
+    expressions = d.parse(
+        f"""
+        MODEL (
+            name db.table,
+            kind FULL,
+        );
+
+        SELECT * FROM db.table
         """
     )
 
     model = load_sql_based_model(expressions)
     with pytest.raises(ConfigError) as ex:
         model.validate_definition()
-    assert error in str(ex.value)
+
+    assert "require inferrable column types" in str(ex.value)
 
 
 def test_model_union_query(sushi_context, assert_exp_eq):
@@ -405,23 +440,6 @@ def get_date(evaluator):
     )
 
 
-def test_model_validation_union_query():
-    expressions = d.parse(
-        """
-        MODEL (
-            name db.table,
-            kind FULL,
-        );
-
-        SELECT a, a UNION SELECT c, c
-        """
-    )
-
-    model = load_sql_based_model(expressions)
-    with pytest.raises(ConfigError, match=r"Found duplicate outer select name 'a'"):
-        model.validate_definition()
-
-
 @use_terminal_console
 def test_model_qualification(tmp_path: Path):
     with patch.object(get_console(), "log_warning") as mock_logger: