Reapply "fix!: dont normalize model meta and lower audit args" (#2559)

This reverts commit c48754f.

Author: tobymao

sqlmesh/core/model/definition.py

@@ -1457,6 +1457,7 @@ def load_sql_based_model(
         python_env=meta_python_env,
         default_catalog=default_catalog,
         quote_identifiers=False,
+        normalize_identifiers=False,
     )
     rendered_meta_exprs = meta_renderer.render()
     if rendered_meta_exprs is None or len(rendered_meta_exprs) != 1:

sqlmesh/core/model/meta.py

@@ -93,7 +93,7 @@ def extract(v: exp.Expression) -> t.Tuple[str, t.Dict[str, exp.Expression]]:
                     raise ConfigError(
                         f"Function '{func}' must be called with key-value arguments like {func}(arg := value)."
                     )
-                kwargs[arg.left.name] = arg.right
+                kwargs[arg.left.name.lower()] = arg.right
             return func.lower(), kwargs
 
         if isinstance(v, (exp.Tuple, exp.Array)):

sqlmesh/core/renderer.py

@@ -48,6 +48,7 @@ def __init__(
         default_catalog: t.Optional[str] = None,
         quote_identifiers: bool = True,
         model_fqn: t.Optional[str] = None,
+        normalize_identifiers: bool = True,
     ):
         self._expression = expression
         self._dialect = dialect
@@ -57,6 +58,7 @@ def __init__(
         self._python_env = python_env or {}
         self._only_execution_time = only_execution_time
         self._default_catalog = default_catalog
+        self._normalize_identifiers = normalize_identifiers
         self._quote_identifiers = quote_identifiers
         self.update_schema({} if schema is None else schema)
         self._cache: t.List[t.Optional[exp.Expression]] = []
@@ -288,9 +290,12 @@ def _expand(node: exp.Expression) -> exp.Expression:
 
     @contextmanager
     def _normalize_and_quote(self, query: E) -> t.Iterator[E]:
-        with d.normalize_and_quote(
-            query, self._dialect, self._default_catalog, quote=self._quote_identifiers
-        ) as query:
+        if self._normalize_identifiers:
+            with d.normalize_and_quote(
+                query, self._dialect, self._default_catalog, quote=self._quote_identifiers
+            ) as query:
+                yield query
+        else:
             yield query
 
     def _should_cache(self, runtime_stage: RuntimeStage, *args: t.Any) -> bool:

tests/core/test_model.py

@@ -2289,14 +2289,19 @@ def test_model_normalization():
         MODEL (
             name foo,
             dialect snowflake,
-            kind INCREMENTAL_BY_UNIQUE_KEY(unique_key a)
+            kind INCREMENTAL_BY_UNIQUE_KEY(unique_key a),
+              audits (
+                not_null(COLUMNS := id)
+            ),
         );
 
         SELECT x.a AS a FROM test.x AS x
         """
     )
     model = SqlModel.parse_raw(load_sql_based_model(expr).json())
     assert model.unique_key == [exp.column("A", quoted=True)]
+    # we should never normalize the model meta, additionally, we should force lower case
+    assert model.audits[0][1] == {"columns": exp.column("id")}
 
     model = create_sql_model(
         "foo",