Fix!: Move linting from load to plan creation

docs/guides/linter.md

@@ -4,7 +4,7 @@
 
 Linting is a powerful tool for improving code quality and consistency. It enables you to automatically validate model definition, ensuring they adhere to your team's best practices.
 
-When a SQLMesh command is executed and the project is loaded, each model's code is checked for compliance with a set of rules you choose.
+When a SQLMesh plan is created, each model's code is checked for compliance with a set of rules you choose.
 
 SQLMesh provides built-in rules, and you can define custom rules. This improves code quality and helps detect issues early in the development cycle when they are simpler to debug.
 

sqlmesh/cli/main.py

@@ -339,6 +339,11 @@ def diff(ctx: click.Context, environment: t.Optional[str] = None) -> None:
     is_flag=True,
     help="Skip tests prior to generating the plan if they are defined.",
 )
+@click.option(
+    "--skip-linter",
+    is_flag=True,
+    help="Skip linting prior to generating the plan if the linter is enabled.",
+)
 @click.option(
     "--restate-model",
     "-r",

sqlmesh/core/console.py

@@ -2194,6 +2194,16 @@ def log_failed_models(self, errors: t.List[NodeExecutionFailedError]) -> None:
 
             self._print("```\n")
 
+    def show_linter_violations(
+        self, violations: t.List[RuleViolation], model: Model, is_error: bool = False
+    ) -> None:
+        severity = "**errors**" if is_error else "warnings"
+        violations_msg = "\n".join(f" - {violation}" for violation in violations)
+        msg = f"\nLinter {severity} for `{model._path}`:\n{violations_msg}\n"
+
+        self._print(msg)
+        self._errors.append(msg)
+
     def log_error(self, message: str) -> None:
         super().log_error(f"```\n\\[ERROR] {message}```\n\n")
 

sqlmesh/core/context.py

@@ -501,8 +501,6 @@ def upsert_model(self, model: t.Union[str, Model], **kwargs: t.Any) -> Model:
 
         model.validate_definition()
 
-        self.lint_models(model)
-
         return model
 
     def scheduler(self, environment: t.Optional[str] = None) -> Scheduler:
@@ -640,8 +638,6 @@ def load(self, update_schemas: bool = True) -> GenericContext[C]:
                 # The model definition can be validated correctly only after the schema is set.
                 model.validate_definition()
 
-            self.lint_models(*models)
-
         duplicates = set(self._models) & set(self._standalone_audits)
         if duplicates:
             raise ConfigError(
@@ -1139,6 +1135,7 @@ def plan(
         no_diff: t.Optional[bool] = None,
         run: bool = False,
         diff_rendered: bool = False,
+        skip_linter: bool = False,
     ) -> Plan:
         """Interactively creates a plan.
 
@@ -1183,6 +1180,7 @@ def plan(
             no_diff: Hide text differences for changed models.
             run: Whether to run latest intervals as part of the plan application.
             diff_rendered: Whether the diff should compare raw vs rendered models
+            skip_linter: Linter runs by default so this will skip it if enabled
 
         Returns:
             The populated Plan object.
@@ -1209,6 +1207,7 @@ def plan(
             enable_preview=enable_preview,
             run=run,
             diff_rendered=diff_rendered,
+            skip_linter=skip_linter,
         )
 
         if no_auto_categorization:
@@ -1250,6 +1249,7 @@ def plan_builder(
         enable_preview: t.Optional[bool] = None,
         run: bool = False,
         diff_rendered: bool = False,
+        skip_linter: bool = False,
     ) -> PlanBuilder:
         """Creates a plan builder.
 
@@ -1305,6 +1305,9 @@ def plan_builder(
         if run and is_dev:
             raise ConfigError("The '--run' flag is only supported for the production environment.")
 
+        if not skip_linter:
+            self.lint_models(*self.models.values())
+
         self._run_plan_tests(skip_tests=skip_tests)
 
         environment_ttl = (

sqlmesh/magics.py

@@ -343,6 +343,11 @@ def test(self, context: Context, line: str, test_def_raw: t.Optional[str] = None
         action="store_true",
         help="Skip the unit tests defined for the model.",
     )
+    @argument(
+        "--skip-linter",
+        action="store_true",
+        help="Skip the linter for the model.",
+    )
     @argument(
         "--restate-model",
         "-r",

tests/cli/test_cli.py

@@ -198,6 +198,29 @@ def test_plan_skip_tests(runner, tmp_path):
     assert_backfill_success(result)
 
 
+def test_plan_skip_linter(runner, tmp_path):
+    create_example_project(tmp_path)
+
+    with open(tmp_path / "config.yaml", "a", encoding="utf-8") as f:
+        f.write(
+            """linter:
+        enabled: True
+        rules: "ALL"
+    """
+        )
+
+    # Successful test run message should not appear with `--skip-tests`
+    # Input: `y` to apply and backfill
+    result = runner.invoke(
+        cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "--skip-linter"], input="y\n"
+    )
+
+    assert result.exit_code == 0
+    assert "Linter warnings" not in result.output
+    assert_new_env(result)
+    assert_backfill_success(result)
+
+
 def test_plan_restate_model(runner, tmp_path):
     create_example_project(tmp_path)
     init_prod_and_backfill(runner, tmp_path)

tests/core/test_context.py

@@ -1698,10 +1698,12 @@ def assert_cached_violations_exist(cache: OptimizedQueryCache, model: Model):
     for query in ("SELECT * FROM tbl", "SELECT t.* FROM tbl"):
         with pytest.raises(LinterError, match=config_err):
             ctx.upsert_model(load_sql_based_model(d.parse(f"MODEL (name test); {query}")))
+            ctx.plan(environment="dev", auto_apply=True, no_prompts=True)
 
-    error_model = load_sql_based_model(d.parse("MODEL (name test); SELECT col"))
+    error_model = load_sql_based_model(d.parse("MODEL (name test); SELECT * FROM (SELECT 1)"))
     with pytest.raises(LinterError, match=config_err):
         ctx.upsert_model(error_model)
+        ctx.plan_builder("dev")
 
     # Case: Ensure error violations are cached if the model did not pass linting
     cache = OptimizedQueryCache(tmp_path / c.CACHE)
@@ -1731,8 +1733,10 @@ def assert_cached_violations_exist(cache: OptimizedQueryCache, model: Model):
                 assert_cached_violations_exist(cache, model2)
 
             ctx.upsert_model(error_model)
+            ctx.plan(environment="dev", auto_apply=True, no_prompts=True)
+
             assert (
-                """Column '"col"' could not be resolved for model '"test"'"""
+                """noselectstar: Query should not contain SELECT * on its outer most projections"""
                 in mock_logger.call_args[0][0]
             )
 
@@ -1779,10 +1783,11 @@ def assert_cached_violations_exist(cache: OptimizedQueryCache, model: Model):
         "MODEL(name test2, audits (at_least_one(column := col)), ignored_rules ['noselectstar']); SELECT * FROM (SELECT 1 AS col);",
     )
 
-    ctx.load()
+    ctx.plan(environment="dev", auto_apply=True, no_prompts=True)
 
     # Case: Ensure we can load & use the user-defined rules
     sushi_context.config.linter = LinterConfig(enabled=True, rules=["aLl"])
+    sushi_context.load()
     sushi_context.upsert_model(
         load_sql_based_model(
             d.parse("MODEL (name sushi.test); SELECT col FROM (SELECT * FROM tbl)"),
@@ -1791,7 +1796,7 @@ def assert_cached_violations_exist(cache: OptimizedQueryCache, model: Model):
     )
 
     with pytest.raises(LinterError, match=config_err):
-        sushi_context.load()
+        sushi_context.plan_builder(environment="dev")
 
     # Case: Ensure the Linter also picks up Python model violations
     @model(name="memory.sushi.model3", is_sql=True, kind="full", dialect="snowflake")
@@ -1813,21 +1818,7 @@ def model4_entrypoint(context, **kwargs):
     for python_model in (model3, model4):
         with pytest.raises(LinterError, match=config_err):
             sushi_context.upsert_model(python_model)
-
-    @model(
-        name="memory.sushi.model5",
-        columns={"col": "int"},
-        owner="test",
-        audits=[("at_least_one", {"column": "col"})],
-    )
-    def model5_entrypoint(context, **kwargs):
-        yield pd.DataFrame({"col": []})
-
-    model5 = model.get_registry()["memory.sushi.model5"].model(
-        module_path=Path("."), path=Path(".")
-    )
-
-    sushi_context.upsert_model(model5)
+            sushi_context.plan(environment="dev", auto_apply=True, no_prompts=True)
 
 
 def test_plan_selector_expression_no_match(sushi_context: Context) -> None:

tests/core/test_integration.py

@@ -4385,12 +4385,10 @@ def test_auto_categorization(sushi_context: Context):
 
 @use_terminal_console
 def test_multi(mocker):
-    context = Context(
-        paths=["examples/multi/repo_1", "examples/multi/repo_2"], gateway="memory", load=False
-    )
+    context = Context(paths=["examples/multi/repo_1", "examples/multi/repo_2"], gateway="memory")
 
     with patch.object(get_console(), "log_warning") as mock_logger:
-        context.load()
+        context.plan_builder(environment="dev")
         warnings = mock_logger.call_args[0][0]
         repo1_path, repo2_path = context.configs.keys()
         assert f"Linter warnings for {repo1_path}" in warnings

tests/core/test_model.py

@@ -301,6 +301,7 @@ def test_model_qualification(tmp_path: Path):
             config=Config(linter=LinterConfig(enabled=True, warn_rules=["ALL"])), paths=tmp_path
         )
         ctx.upsert_model(load_sql_based_model(expressions))
+        ctx.plan_builder("dev")
 
         assert (
             """Column '"a"' could not be resolved for model '"db"."table"', the column may not exist or is ambiguous."""
@@ -327,6 +328,7 @@ def test_model_missing_audits(tmp_path: Path):
             paths=tmp_path,
         )
         ctx.upsert_model(load_sql_based_model(expressions))
+        ctx.plan_builder("plan")
 
         assert (
             """Model `audits` must be configured to test data quality."""
@@ -2841,6 +2843,7 @@ def test_update_schema(tmp_path: Path):
     )
     with patch.object(get_console(), "log_warning") as mock_logger:
         ctx.upsert_model(model)
+        ctx.plan_builder("dev")
         assert (
             missing_schema_warning_msg('"db"."table"', ('"table_b"',))
             in mock_logger.call_args[0][0]
@@ -2895,13 +2898,14 @@ def test_missing_schema_warnings(tmp_path: Path):
         model = load_sql_based_model(d.parse("MODEL (name test); SELECT * FROM a CROSS JOIN b"))
         model.update_schema(partial_schema)
         ctx.upsert_model(model)
-
+        ctx.plan_builder("dev")
         assert missing_schema_warning_msg('"test"', ('"b"',)) in mock_logger.call_args[0][0]
 
     # star, no schema
     with patch.object(console, "log_warning") as mock_logger:
         model = load_sql_based_model(d.parse("MODEL (name test); SELECT * FROM b JOIN a"))
         ctx.upsert_model(model)
+        ctx.plan_builder("dev")
         assert missing_schema_warning_msg('"test"', ('"a"', '"b"')) in mock_logger.call_args[0][0]
 
     # no star, full schema
@@ -8062,16 +8066,18 @@ def model_with_virtual_statements(context, **kwargs):
 
 def test_compile_time_checks(tmp_path: Path):
     ctx = Context(
-        config=Config(model_defaults=ModelDefaultsConfig(dialect="duckdb")), paths=tmp_path
+        config=Config(
+            model_defaults=ModelDefaultsConfig(dialect="duckdb"),
+            linter=LinterConfig(
+                enabled=True, rules=["ambiguousorinvalidcolumn", "invalidselectstarexpansion"]
+            ),
+        ),
+        paths=tmp_path,
     )
 
     cfg_err = "Linter detected errors in the code. Please fix them before proceeding."
 
     # Strict SELECT * expansion
-    linter_cfg = LinterConfig(
-        enabled=True, rules=["ambiguousorinvalidcolumn", "invalidselectstarexpansion"]
-    )
-    ctx.config.linter = linter_cfg
     strict_query = d.parse(
         """
     MODEL (
@@ -8082,10 +8088,9 @@ def test_compile_time_checks(tmp_path: Path):
     """
     )
 
-    ctx.load()
-
     with pytest.raises(LinterError, match=cfg_err):
         ctx.upsert_model(load_sql_based_model(strict_query))
+        ctx.plan_builder("dev")
 
     # Strict column resolution
     strict_query = d.parse(
@@ -8100,6 +8105,7 @@ def test_compile_time_checks(tmp_path: Path):
 
     with pytest.raises(LinterError, match=cfg_err):
         ctx.upsert_model(load_sql_based_model(strict_query))
+        ctx.plan_builder("dev")
 
 
 def test_partition_interval_unit():