Feat!: Add 'validate_query' model attribute for compile-time checks (#3614)

Author: VaggelisD

docs/concepts/models/overview.md

@@ -443,6 +443,9 @@ to `false` causes SQLMesh to disable query canonicalization & simplification. Th
 !!! warning
     Turning off the optimizer may prevent column-level lineage from working for the affected model and its descendants, unless all columns in the model's query are qualified and it contains no star projections (e.g. `SELECT *`).
 
+### validate_query
+:   Whether the model's query will be validated at compile time. This attribute is `false` by default. Setting it to `true` causes SQLMesh to raise an error instead of emitting warnings. This will display invalid columns in your SQL statements along with models containing `SELECT *` that cannot be automatically expanded to list out all columns. This ensures SQL is verified locally before time and money are spent running the SQL in your data warehouse.
+
 ## Incremental Model Properties
 
 These properties can be specified in an incremental model's `kind` definition.

docs/reference/model_configuration.md

@@ -39,7 +39,7 @@ Configuration options for SQLMesh model properties. Supported by all model kinds
 | `enabled`             | Whether the model is enabled. This attribute is `true` by default. Setting it to `false` causes SQLMesh to ignore this model when loading the project.                                                                                                                                                                                                      |       bool        |    N     |
 | `gateway`             | Specifies the gateway to use for the execution of this model. When not specified, the default gateway is used.                                                                                                                                                                                                       |       str        |    N     |
 | `optimize_query`             | Whether the model's query should be optimized. This attribute is `true` by default. Setting it to `false` causes SQLMesh to disable query canonicalization & simplification. This should be turned off only if the optimized query leads to errors such as surpassing text limit.                                                                                                                                                                                                      |       bool        |    N     |
-
+| `validate_query`             | `validate_query`             | Whether the model's query will be strictly validated at compile time. This attribute is `false` by default. Setting it to `true` causes SQLMesh to raise an error instead of emitting warnings. This will display invalid columns in your SQL statements along with models containing `SELECT *` that cannot be automatically expanded to list out all columns.                                                                                                                                                                                              |       bool        |    N     |
 ### Model defaults
 
 The SQLMesh project-level configuration must contain the `model_defaults` key and must specify a value for its `dialect` key. Other values are set automatically unless explicitly overridden in the model definition. Learn more about project-level configuration in the [configuration guide](../guides/configuration.md).

sqlmesh/core/config/model.py

@@ -47,6 +47,7 @@ class ModelDefaultsConfig(BaseConfig):
     session_properties: t.Optional[t.Dict[str, t.Any]] = None
     audits: t.Optional[t.List[FunctionCall]] = None
     optimize_query: t.Optional[bool] = None
+    validate_query: t.Optional[bool] = None
 
     _model_kind_validator = model_kind_validator
     _on_destructive_change_validator = on_destructive_change_validator

sqlmesh/core/model/cache.py

@@ -135,6 +135,7 @@ def _entry_name(model: SqlModel) -> str:
         hash_data.append(str([gen(d) for d in model.macro_definitions]))
         hash_data.append(str([(k, v) for k, v in model.sorted_python_env]))
         hash_data.extend(model.jinja_macros.data_hash_values)
+        hash_data.extend(str(model.validate_query))
         return f"{model.name}_{crc32(hash_data)}"
 
 

sqlmesh/core/model/common.py

@@ -303,6 +303,7 @@ def depends_on(cls: t.Type, v: t.Any, values: t.Dict[str, t.Any]) -> t.Optional[
     "allow_partials",
     "enabled",
     "optimize_query",
+    "validate_query",
     mode="before",
     check_fields=False,
 )(parse_bool)

sqlmesh/core/model/definition.py

@@ -221,6 +221,7 @@ def render_definition(
                     "enabled",
                     "inline_audits",
                     "optimize_query",
+                    "validate_query",
                 ):
                     expressions.append(
                         exp.Property(
@@ -912,11 +913,19 @@ def validate_definition(self) -> None:
                 self._path,
             )
 
-        if not self.is_sql and self.optimize_query is not None:
-            raise_config_error(
-                "SQLMesh query optimizer can only be enabled/disabled for SQL models",
-                self._path,
-            )
+        # The following attributes should be set only for SQL models
+        if not self.is_sql:
+            if self.optimize_query:
+                raise_config_error(
+                    "SQLMesh query optimizer can only be enabled for SQL models",
+                    self._path,
+                )
+
+            if self.validate_query:
+                raise_config_error(
+                    "Query validation can only be enabled for SQL models",
+                    self._path,
+                )
 
     def is_breaking_change(self, previous: Model) -> t.Optional[bool]:
         """Determines whether this model is a breaking change in relation to the `previous` model.
@@ -1013,6 +1022,7 @@ def metadata_hash(self) -> str:
                 self.project,
                 str(self.allow_partials),
                 gen(self.session_properties_) if self.session_properties_ else None,
+                str(self.validate_query) if self.validate_query is not None else None,
             ]
 
             for audit_name, audit_args in sorted(self.audits, key=lambda a: a[0]):
@@ -1354,6 +1364,7 @@ def _query_renderer(self) -> QueryRenderer:
             default_catalog=self.default_catalog,
             quote_identifiers=not no_quote_identifiers,
             optimize_query=self.optimize_query,
+            validate_query=self.validate_query,
         )
 
     @property

sqlmesh/core/model/meta.py

@@ -79,6 +79,7 @@ class ModelMeta(_Node):
     physical_version: t.Optional[str] = None
     gateway: t.Optional[str] = None
     optimize_query: t.Optional[bool] = None
+    validate_query: t.Optional[bool] = None
 
     _bool_validator = bool_validator
     _model_kind_validator = model_kind_validator

sqlmesh/core/renderer.py

@@ -52,6 +52,7 @@ def __init__(
         model_fqn: t.Optional[str] = None,
         normalize_identifiers: bool = True,
         optimize_query: t.Optional[bool] = True,
+        validate_query: t.Optional[bool] = False,
     ):
         self._expression = expression
         self._dialect = dialect
@@ -67,6 +68,7 @@ def __init__(
         self._cache: t.List[t.Optional[exp.Expression]] = []
         self._model_fqn = model_fqn
         self._optimize_query_flag = optimize_query is not False
+        self._validate_query = validate_query
 
     def update_schema(self, schema: t.Dict[str, t.Any]) -> None:
         self.schema = d.normalize_mapping_schema(schema, dialect=self._dialect)
@@ -526,12 +528,17 @@ def _optimize_query(self, query: exp.Query, all_deps: t.Set[str]) -> exp.Query:
         if self._model_fqn and not should_optimize and any(s.is_star for s in query.selects):
             deps = ", ".join(f"'{dep}'" for dep in sorted(missing_deps))
 
-            logger.warning(
+            warning = (
                 f"SELECT * cannot be expanded due to missing schema(s) for model(s): {deps}. "
                 "Run `sqlmesh create_external_models` and / or make sure that the model "
-                f"'{self._model_fqn}' can be rendered at parse time.",
+                f"'{self._model_fqn}' can be rendered at parse time."
             )
 
+            if self._validate_query:
+                raise_config_error(warning, self._path)
+
+            logger.warning(warning)
+
         try:
             if should_optimize:
                 query = query.copy()
@@ -549,11 +556,16 @@ def _optimize_query(self, query: exp.Query, all_deps: t.Set[str]) -> exp.Query:
                     )
                 )
         except SqlglotError as ex:
+            warning = (
+                f"{ex} for model '{self._model_fqn}', the column may not exist or is ambiguous"
+            )
+
+            if self._validate_query:
+                raise_config_error(warning, self._path)
+
             query = original
 
-            logger.warning(
-                "%s for model '%s', the column may not exist or is ambiguous", ex, self._model_fqn
-            )
+            logger.warning(warning)
         except Exception as ex:
             raise_config_error(
                 f"Failed to optimize query, please file an issue at https://github.com/TobikoData/sqlmesh/issues/new. {ex}",

tests/core/test_model.py

@@ -291,7 +291,7 @@ def test_model_qualification():
         model.render_query(needs_optimization=True)
         assert (
             mock_logger.call_args[0][0]
-            == "%s for model '%s', the column may not exist or is ambiguous"
+            == """Column '"a"' could not be resolved for model '"db"."table"', the column may not exist or is ambiguous"""
         )
 
 
@@ -1001,10 +1001,10 @@ def test_seed_on_virtual_update_statements():
         CREATE TABLE x{{ 1 + 1 }};
         JINJA_END;
 
-        ON_VIRTUAL_UPDATE_BEGIN; 
-        JINJA_STATEMENT_BEGIN;     
+        ON_VIRTUAL_UPDATE_BEGIN;
+        JINJA_STATEMENT_BEGIN;
         GRANT SELECT ON VIEW {{ this_model }} TO ROLE dev_role;
-        JINJA_END;  
+        JINJA_END;
         DROP TABLE x2;
         ON_VIRTUAL_UPDATE_END;
 
@@ -6595,24 +6595,29 @@ def test_model_optimize(tmp_path: Path, assert_exp_eq):
         context.plan(auto_apply=True, no_prompts=True)
 
     # Ensure non-SQLModels raise if optimize_query is not None
+    seed_path = tmp_path / "seed.csv"
+    model_kind = SeedKind(path=str(seed_path.absolute()))
+    with open(seed_path, "w", encoding="utf-8") as fd:
+        fd.write(
+            """
+col_a,col_b,col_c
+1,text_a,1.0
+2,text_b,2.0"""
+        )
+    model = create_seed_model("test_db.test_seed_model", model_kind, optimize_query=True)
+    context = Context(config=Config(model_defaults=ModelDefaultsConfig(dialect="duckdb")))
+
     with pytest.raises(
         ConfigError,
-        match=r"SQLMesh query optimizer can only be enabled/disabled for SQL models",
+        match=r"SQLMesh query optimizer can only be enabled for SQL models",
     ):
-        seed_path = tmp_path / "seed.csv"
-        model_kind = SeedKind(path=str(seed_path.absolute()))
-        with open(seed_path, "w", encoding="utf-8") as fd:
-            fd.write(
-                """
-    col_a,col_b,col_c
-    1,text_a,1.0
-    2,text_b,2.0"""
-            )
-        model = create_seed_model("test_db.test_seed_model", model_kind, optimize_query=True)
-        context = Context(config=Config(model_defaults=ModelDefaultsConfig(dialect="duckdb")))
         context.upsert_model(model)
         context.plan(auto_apply=True, no_prompts=True)
 
+    model = create_seed_model("test_db.test_seed_model", model_kind, optimize_query=False)
+    context.upsert_model(model)
+    context.plan(auto_apply=True, no_prompts=True)
+
 
 def test_column_description_metadata_change():
     context = Context(config=Config())
@@ -6762,15 +6767,15 @@ def test_model_on_virtual_update(make_snapshot: t.Callable):
     def resolve_parent_name(evaluator, name):
         return evaluator.resolve_table(name.name)
 
-    virtual_update_statements = """ 
+    virtual_update_statements = """
         CREATE OR REPLACE VIEW test_view FROM demo_db.table;
         GRANT SELECT ON VIEW @this_model TO ROLE owner_name;
         JINJA_STATEMENT_BEGIN;
         GRANT SELECT ON VIEW {{this_model}} TO ROLE admin;
         JINJA_END;
         GRANT REFERENCES, SELECT ON FUTURE VIEWS IN DATABASE demo_db TO ROLE owner_name;
         @resolve_parent_name('parent');
-        GRANT SELECT ON VIEW demo_db.table /* sqlglot.meta replace=false */ TO ROLE admin; 
+        GRANT SELECT ON VIEW demo_db.table /* sqlglot.meta replace=false */ TO ROLE admin;
     """
 
     expressions = d.parse(
@@ -6784,7 +6789,7 @@ def resolve_parent_name(evaluator, name):
 
         on_virtual_update_begin;
 
-        {virtual_update_statements} 
+        {virtual_update_statements}
 
         on_virtual_update_end;
 
@@ -6918,3 +6923,111 @@ def model_with_virtual_statements(context, **kwargs):
         rendered_statements[2].sql()
         == "GRANT REFERENCES, SELECT ON FUTURE VIEWS IN DATABASE db TO ROLE dev_role"
     )
+
+
+def test_compile_time_checks(tmp_path: Path, assert_exp_eq):
+    # Strict SELECT * expansion
+    strict_query = d.parse(
+        """
+    MODEL (
+        name test,
+        validate_query True,
+    );
+
+    SELECT * FROM tbl
+    """
+    )
+
+    with pytest.raises(
+        ConfigError,
+        match=r".*cannot be expanded due to missing schema.*",
+    ):
+        load_sql_based_model(strict_query).render_query()
+
+    # Strict column resolution
+    strict_query = d.parse(
+        """
+    MODEL (
+        name test,
+        validate_query True,
+    );
+
+    SELECT foo
+    """
+    )
+
+    with pytest.raises(
+        ConfigError,
+        match=r"""Column '"foo"' could not be resolved for model.*""",
+    ):
+        load_sql_based_model(strict_query).render_query()
+
+    # Non-strict model with strict defaults raises error, otherwise can still render
+    strict_default = ModelDefaultsConfig(validate_query=True).dict()
+    query = d.parse(
+        """
+    MODEL (
+        name test,
+    );
+
+    SELECT * FROM tbl
+    """
+    )
+
+    with pytest.raises(
+        ConfigError,
+        match=r".*cannot be expanded due to missing schema.*",
+    ):
+        load_sql_based_model(query, defaults=strict_default).render_query()
+
+    assert_exp_eq(load_sql_based_model(query).render_query(), 'SELECT * FROM "tbl" AS "tbl"')
+
+    # Ensure plan works for valid queries & cache is invalidated if strict changes
+    context = Context(config=Config(model_defaults=ModelDefaultsConfig(dialect="duckdb")))
+
+    query = d.parse(
+        """
+    MODEL (
+        name db.test,
+        validate_query True,
+    );
+
+    SELECT 1 AS col
+    """
+    )
+
+    context.upsert_model(load_sql_based_model(query, default_catalog=context.default_catalog))
+    context.plan(auto_apply=True, no_prompts=True)
+
+    context.upsert_model("db.test", validate_query=False)
+    plan = context.plan(no_prompts=True, auto_apply=True)
+
+    snapshots = list(plan.snapshots.values())
+    assert len(snapshots) == 1
+
+    snapshot = snapshots[0]
+    assert len(snapshot.previous_versions) == 1
+    assert snapshot.change_category == SnapshotChangeCategory.METADATA
+
+    # Ensure non-SQLModels raise if strict mode is set to True
+    seed_path = tmp_path / "seed.csv"
+    model_kind = SeedKind(path=str(seed_path.absolute()))
+    with open(seed_path, "w", encoding="utf-8") as fd:
+        fd.write(
+            """
+col_a,col_b,col_c
+1,text_a,1.0"""
+        )
+    model = create_seed_model("test_db.test_seed_model", model_kind, validate_query=True)
+    context = Context(config=Config(model_defaults=ModelDefaultsConfig(dialect="duckdb")))
+
+    with pytest.raises(
+        ConfigError,
+        match=r"Query validation can only be enabled for SQL models at",
+    ):
+        context.upsert_model(model)
+        context.plan(auto_apply=True, no_prompts=True)
+
+    model = create_seed_model("test_db.test_seed_model", model_kind, validate_query=False)
+    context.upsert_model(model)
+    context.plan(auto_apply=True, no_prompts=True)

tests/core/test_snapshot.py

@@ -760,7 +760,7 @@ def test_fingerprint(model: Model, parent_model: Model):
 
     original_fingerprint = SnapshotFingerprint(
         data_hash="1312415267",
-        metadata_hash="2573378960",
+        metadata_hash="2476734280",
     )
 
     assert fingerprint == original_fingerprint
@@ -819,7 +819,7 @@ def test_fingerprint_seed_model():
 
     expected_fingerprint = SnapshotFingerprint(
         data_hash="1909791099",
-        metadata_hash="3403817841",
+        metadata_hash="1153541408",
     )
 
     model = load_sql_based_model(expressions, path=Path("./examples/sushi/models/test_model.sql"))
@@ -858,7 +858,7 @@ def test_fingerprint_jinja_macros(model: Model):
     )
     original_fingerprint = SnapshotFingerprint(
         data_hash="923305614",
-        metadata_hash="2573378960",
+        metadata_hash="2476734280",
     )
 
     fingerprint = fingerprint_from_node(model, nodes={})