Repository: TobikoData/sqlmesh
Pull request #1778 (2023-11-30 19:51:30)

feat: improve logging
previously, it was a common flow to have users rerun plans with --debug in order for the core team to get more comprehensive logs. this pr trims down logs and also logs all relevant sql queries as info to the file so that by default we log essential information for debugging.


Patch:
diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md
index 150d958d..a60a27fc 100644
--- a/docs/reference/configuration.md
+++ b/docs/reference/configuration.md
@@ -34,7 +34,8 @@ Configuration options for SQLMesh environment creation and promotion.
 | `auto_categorize_changes`    | Indicates whether SQLMesh should attempt to automatically [categorize](../concepts/plans.md#change-categories) model changes during plan creation per each model source type ([additional details](../guides/configuration.md#auto-categorize-changes))                                                                      | dict[string, string] | N        |
 | `default_target_environment` | The name of the environment that will be the default target for the `sqlmesh plan` and `sqlmesh run` commands. (Default: `prod`)                                                                                                                                                                   | string               | N        |
 | `physical_schema_override`   | A mapping from model schema names to names of schemas in which physical tables for the corresponding models will be placed - [addition details](../guides/configuration.md#physical-schema-override). (Default physical schema name: `sqlmesh__[model schema]`)                                                                                                                                                                   | string               | N        |
-| `environment_suffix_target`   | Whether SQLMesh views should append their environment name to the `schema` or `table` - [additional details](../guides/configuration.md#view-schema-override). (Default: `schema`)                                                                                                                                                                   | string               | N        |
+| `environment_suffix_target`  | Whether SQLMesh views should append their environment name to the `schema` or `table` - [additional details](../guides/configuration.md#view-schema-override). (Default: `schema`)                                                                                                                                                                   | string               | N        |
+| `log_limit`                  | The default number of logs to keep (Default: `20`) | int               | N        |
 
 ### Model defaults
 
diff --git a/sqlmesh/__init__.py b/sqlmesh/__init__.py
index adfbf734..44aa1b8d 100644
--- a/sqlmesh/__init__.py
+++ b/sqlmesh/__init__.py
@@ -3,6 +3,7 @@
 """
 from __future__ import annotations
 
+import glob
 import logging
 import os
 import sys
@@ -14,6 +15,7 @@ from sqlmesh.core.dialect import extend_sqlglot
 
 extend_sqlglot()
 
+from sqlmesh.core import constants as c
 from sqlmesh.core.config import Config
 from sqlmesh.core.context import Context, ExecutionContext
 from sqlmesh.core.engine_adapter import EngineAdapter
@@ -90,6 +92,7 @@ if runtime_env.is_notebook:
 
 
 LOG_FORMAT = "%(asctime)s - %(threadName)s - %(name)s - %(levelname)s - %(message)s (%(filename)s:%(lineno)d)"
+LOG_PREFIX = "logs/sqlmesh_"
 
 
 # SO: https://stackoverflow.com/questions/384076/how-can-i-color-python-logging-output
@@ -121,10 +124,37 @@ def configure_logging(
     ignore_warnings: bool = False,
     write_to_stdout: bool = True,
     write_to_file: bool = True,
+    log_limit: int = c.DEFAULT_LOG_LIMIT,
 ) -> None:
     logger = logging.getLogger()
-
     debug = force_debug or debug_mode_enabled()
+
+    if not logger.hasHandlers():
+        # base logger needs to be the lowest level that we plan to log
+        level = logging.DEBUG if debug else logging.INFO
+        logger.setLevel(level)
+
+        stdout_handler = logging.StreamHandler(sys.stdout)
+        stdout_handler.setFormatter(CustomFormatter())
+        stdout_handler.setLevel(
+            level if write_to_stdout else (logging.ERROR if ignore_warnings else logging.WARNING)
+        )
+        logger.addHandler(stdout_handler)
+
+        if write_to_file:
+            os.makedirs("logs", exist_ok=True)
+            filename = f"{LOG_PREFIX}{datetime.now().strftime('%Y_%m_%d_%H_%M_%S')}.log"
+            file_handler = logging.FileHandler(filename, mode="w", encoding="utf-8")
+            # the log files should always log at least info so that users will always have
+            # minimal info for debugging even if they specify "ignore_warnings"
+            file_handler.setLevel(level)
+            file_handler.setFormatter(logging.Formatter(LOG_FORMAT))
+            logger.addHandler(file_handler)
+
+    if log_limit > 0:
+        for path in list(sorted(glob.glob("{LOG_PREFIX}*.log"), reverse=True))[log_limit:]:
+            os.remove(path)
+
     if debug:
         import faulthandler
         import signal
@@ -136,22 +166,3 @@ def configure_logging(
         # Windows doesn't support register so we check for it here
         if hasattr(faulthandler, "register"):
             faulthandler.register(signal.SIGUSR1.value)
-
-        level = logging.DEBUG
-        logger.setLevel(level)
-        if not logger.hasHandlers():
-            if write_to_stdout:
-                handler = logging.StreamHandler(sys.stdout)
-                handler.setLevel(level)
-                handler.setFormatter(CustomFormatter())
-                logger.addHandler(handler)
-
-            if write_to_file:
-                os.makedirs("logs", exist_ok=True)
-                filename = f"logs/sqlmesh_{datetime.now().strftime('%Y_%m_%d_%H_%M_%S')}.log"
-                file_handler = logging.FileHandler(filename, mode="w", encoding="utf-8")
-                file_handler.setLevel(level)
-                file_handler.setFormatter(logging.Formatter(LOG_FORMAT))
-                logger.addHandler(file_handler)
-    elif ignore_warnings:
-        logger.setLevel(logging.ERROR)
diff --git a/sqlmesh/cli/main.py b/sqlmesh/cli/main.py
index 2ddbe02a..1f8adf11 100644
--- a/sqlmesh/cli/main.py
+++ b/sqlmesh/cli/main.py
@@ -11,6 +11,7 @@ from sqlmesh import configure_logging
 from sqlmesh.cli import error_handler
 from sqlmesh.cli import options as opt
 from sqlmesh.cli.example_project import ProjectTemplate, init_example_project
+from sqlmesh.core.config import load_configs
 from sqlmesh.core.context import Context
 from sqlmesh.utils.date import TimeLike
 from sqlmesh.utils.errors import MissingDependencyError
@@ -74,12 +75,14 @@ def cli(
     if "--help" in sys.argv:
         return
 
-    configure_logging(debug, ignore_warnings, write_to_stdout=False)
+    configs = load_configs(config, paths)
+    log_limit = list(configs.values())[0].log_limit
+    configure_logging(debug, ignore_warnings, write_to_stdout=False, log_limit=log_limit)
 
     try:
         context = Context(
             paths=paths,
-            config=config,
+            config=configs,
             gateway=gateway,
             load=load,
         )
diff --git a/sqlmesh/core/config/__init__.py b/sqlmesh/core/config/__init__.py
index bd2bd883..456f3043 100644
--- a/sqlmesh/core/config/__init__.py
+++ b/sqlmesh/core/config/__init__.py
@@ -15,7 +15,11 @@ from sqlmesh.core.config.connection import (
     parse_connection_config,
 )
 from sqlmesh.core.config.gateway import GatewayConfig
-from sqlmesh.core.config.loader import load_config_from_paths, load_config_from_yaml
+from sqlmesh.core.config.loader import (
+    load_config_from_paths,
+    load_config_from_yaml,
+    load_configs,
+)
 from sqlmesh.core.config.model import ModelDefaultsConfig
 from sqlmesh.core.config.root import Config
 from sqlmesh.core.config.run import RunConfig
diff --git a/sqlmesh/core/config/common.py b/sqlmesh/core/config/common.py
index 72409f6c..d035700c 100644
--- a/sqlmesh/core/config/common.py
+++ b/sqlmesh/core/config/common.py
@@ -24,6 +24,9 @@ class EnvironmentSuffixTarget(str, Enum):
     def default(cls) -> EnvironmentSuffixTarget:
         return EnvironmentSuffixTarget.SCHEMA
 
+    def __repr__(self) -> str:
+        return str(self)
+
 
 def _concurrent_tasks_validator(v: t.Any) -> int:
     if isinstance(v, str):
diff --git a/sqlmesh/core/config/loader.py b/sqlmesh/core/config/loader.py
index 78a7cc0f..76978c76 100644
--- a/sqlmesh/core/config/loader.py
+++ b/sqlmesh/core/config/loader.py
@@ -4,14 +4,52 @@ import os
 import typing as t
 from pathlib import Path
 
+from sqlglot.helper import ensure_list
+
 from sqlmesh.core import constants as c
 from sqlmesh.core.config.root import Config
-from sqlmesh.utils import merge_dicts, sys_path
+from sqlmesh.utils import env_vars, merge_dicts, sys_path
 from sqlmesh.utils.errors import ConfigError
 from sqlmesh.utils.metaprogramming import import_python_file
 from sqlmesh.utils.yaml import load as yaml_load
 
 
+def load_configs(
+    config: t.Optional[str | Config],
+    paths: t.Union[str | Path, t.Iterable[str | Path]],
+    sqlmesh_path: Path = c.SQLMESH_PATH,
+) -> t.Dict[Path, Config]:
+    config = config or "config"
+
+    absolute_paths = [
+        Path(t.cast(t.Union[str, Path], path)).absolute() for path in ensure_list(paths)
+    ]
+
+    if isinstance(config, Config):
+        return {path: config for path in absolute_paths}
+
+    config_env_vars = None
+    personal_paths = [
+        sqlmesh_path / "config.yml",
+        sqlmesh_path / "config.yaml",
+    ]
+    for path in personal_paths:
+        if path.exists():
+            config_env_vars = load_config_from_yaml(path).get("env_vars")
+            if config_env_vars:
+                break
+
+    with env_vars(config_env_vars if config_env_vars else {}):
+        return {
+            path: load_config_from_paths(
+                project_paths=[path / "config.py", path / "config.yml", path / "config.yaml"],
+                personal_paths=personal_paths,
+                config_name=config,
+            )
+            for path in absolute_paths
+        }
+
+
 def load_config_from_paths(
     project_paths: t.Optional[t.List[Path]] = None,
     personal_paths: t.Optional[t.List[Path]] = None,
diff --git a/sqlmesh/core/config/root.py b/sqlmesh/core/config/root.py
index 8ea9484e..b2fca1ee 100644
--- a/sqlmesh/core/config/root.py
+++ b/sqlmesh/core/config/root.py
@@ -61,6 +61,7 @@ class Config(BaseConfig):
         physical_schema_override: A mapping from model schema names to names of schemas in which physical tables for corresponding models will be placed.
         environment_suffix_target: Indicates whether to append the environment name to the schema or table name.
         default_target_environment: The name of the environment that will be the default target for the `sqlmesh plan` and `sqlmesh run` commands.
+        log_limit: The default number of logs to keep.
     """
 
     gateways: t.Dict[str, GatewayConfig] = {"": GatewayConfig()}
@@ -88,6 +89,7 @@ class Config(BaseConfig):
         default=EnvironmentSuffixTarget.default
     )
     default_target_environment: str = c.PROD
+    log_limit: int = c.DEFAULT_LOG_LIMIT
     cicd_bot: t.Optional[CICDBotConfig] = None
     run: RunConfig = RunConfig()
 
diff --git a/sqlmesh/core/constants.py b/sqlmesh/core/constants.py
index 97d88137..f5e1be2c 100644
--- a/sqlmesh/core/constants.py
+++ b/sqlmesh/core/constants.py
@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 import datetime
+from pathlib import Path
 
 SQLMESH = "sqlmesh"
+SQLMESH_PATH = Path.home() / ".sqlmesh"
 
 PROD = "prod"
 """Prod"""
@@ -31,6 +33,9 @@ EPOCH = datetime.date(1970, 1, 1)
 DEFAULT_MAX_LIMIT = 1000
 """The default maximum row limit that is used when evaluating a model."""
 
+DEFAULT_LOG_LIMIT = 20
+"""The default number of logs to keep."""
+
 AUDITS = "audits"
 MACROS = "macros"
 METRICS = "metrics"
diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py
index 47edb11a..f3239b84 100644
--- a/sqlmesh/core/context.py
+++ b/sqlmesh/core/context.py
@@ -50,12 +50,7 @@ from sqlglot import exp
 
 from sqlmesh.core import constants as c
 from sqlmesh.core.audit import Audit, StandaloneAudit
-from sqlmesh.core.config import (
-    CategorizerConfig,
-    Config,
-    load_config_from_paths,
-    load_config_from_yaml,
-)
+from sqlmesh.core.config import CategorizerConfig, Config, load_configs
 from sqlmesh.core.console import Console, get_console
 from sqlmesh.core.context_diff import ContextDiff
 from sqlmesh.core.dialect import (
@@ -101,7 +96,7 @@ from sqlmesh.core.test import (
     run_tests,
 )
 from sqlmesh.core.user import User
-from sqlmesh.utils import UniqueKeyDict, env_vars, sys_path
+from sqlmesh.utils import UniqueKeyDict, sys_path
 from sqlmesh.utils.dag import DAG
 from sqlmesh.utils.date import TimeLike, now_ds, to_date
 from sqlmesh.utils.errors import (
@@ -250,20 +245,7 @@ class Context(BaseContext):
         users: t.Optional[t.List[User]] = None,
     ):
         self.console = console or get_console()
-
-        self.sqlmesh_path = Path.home() / ".sqlmesh"
-
-        if isinstance(config, dict):
-            self.configs = config
-        else:
-            self.configs = self._load_configs(
-                config or "config",
-                [
-                    Path(path).absolute()
-                    for path in ([paths] if isinstance(paths, str) else list(paths))
-                ],
-            )
-
+        self.configs = config if isinstance(config, dict) else load_configs(config, paths)
         self.dag: DAG[str] = DAG()
         self._models: UniqueKeyDict[str, Model] = UniqueKeyDict("models")
         self._audits: UniqueKeyDict[str, Audit] = UniqueKeyDict("audits")
@@ -314,7 +296,7 @@ class Context(BaseContext):
     def snapshot_evaluator(self) -> SnapshotEvaluator:
         if not self._snapshot_evaluator:
             self._snapshot_evaluator = SnapshotEvaluator(
-                self.engine_adapter,
+                self.engine_adapter.with_log_level(logging.INFO),
                 ddl_concurrent_tasks=self.concurrent_tasks,
                 console=self.console,
             )
@@ -1476,33 +1458,6 @@ class Context(BaseContext):
             state_reader=self.state_reader,
         )
 
-    def _load_configs(
-        self, config: t.Union[str, Config], paths: t.List[Path]
-    ) -> t.Dict[Path, Config]:
-        if isinstance(config, Config):
-            return {path: config for path in paths}
-
-        config_env_vars = None
-        personal_paths = [
-            self.sqlmesh_path / "config.yml",
-            self.sqlmesh_path / "config.yaml",
-        ]
-        for path in personal_paths:
-            if path.exists():
-                config_env_vars = load_config_from_yaml(path).get("env_vars")
-                if config_env_vars:
-                    break
-
-        with env_vars(config_env_vars if config_env_vars else {}):
-            return {
-                path: load_config_from_paths(
-                    project_paths=[path / "config.py", path / "config.yml", path / "config.yaml"],
-                    personal_paths=personal_paths,
-                    config_name=config,
-                )
-                for path in paths
-            }
-
     def _run_janitor(self) -> None:
         expired_environments = self.state_sync.delete_expired_environments()
         cleanup_expired_views(self.engine_adapter, expired_environments)
diff --git a/sqlmesh/core/engine_adapter/__init__.py b/sqlmesh/core/engine_adapter/__init__.py
index 4a145da4..f34bffe2 100644
--- a/sqlmesh/core/engine_adapter/__init__.py
+++ b/sqlmesh/core/engine_adapter/__init__.py
@@ -36,26 +36,17 @@ DIALECT_ALIASES = {
 
 
 def create_engine_adapter(
-    connection_factory: t.Callable[[], t.Any],
-    dialect: str,
-    multithreaded: bool = False,
-    **kwargs: t.Any,
+    connection_factory: t.Callable[[], t.Any], dialect: str, **kwargs: t.Any
 ) -> EngineAdapter:
     dialect = dialect.lower()
     dialect = DIALECT_ALIASES.get(dialect, dialect)
     engine_adapter = DIALECT_TO_ENGINE_ADAPTER.get(dialect)
     if engine_adapter is None:
-        return EngineAdapter(
-            connection_factory,
-            dialect,
-            multithreaded=multithreaded,
-            **kwargs,
-        )
+        return EngineAdapter(connection_factory, dialect, **kwargs)
     if engine_adapter is EngineAdapterWithIndexSupport:
         return EngineAdapterWithIndexSupport(
             connection_factory,
             dialect,
-            multithreaded=multithreaded,
             **kwargs,
         )
-    return engine_adapter(connection_factory, multithreaded=multithreaded, **kwargs)
+    return engine_adapter(connection_factory, **kwargs)
diff --git a/sqlmesh/core/engine_adapter/base.py b/sqlmesh/core/engine_adapter/base.py
index 893c4a10..69ae13aa 100644
--- a/sqlmesh/core/engine_adapter/base.py
+++ b/sqlmesh/core/engine_adapter/base.py
@@ -173,6 +173,7 @@ class EngineAdapter:
         cursor_kwargs: t.Optional[t.Dict[str, t.Any]] = None,
         cursor_init: t.Optional[t.Callable[[t.Any], None]] = None,
         default_catalog: t.Optional[str] = None,
+        execute_log_level: int = logging.DEBUG,
         **kwargs: t.Any,
     ):
         self.dialect = dialect.lower() or self.DIALECT
@@ -181,8 +182,23 @@ class EngineAdapter:
         )
         self.sql_gen_kwargs = sql_gen_kwargs or {}
         self.default_catalog = default_catalog
+        self._execute_log_level = execute_log_level
         self._extra_config = kwargs
 
+    def with_log_level(self, level: int) -> EngineAdapter:
+        adapter = self.__class__(
+            lambda: None,
+            dialect=self.dialect,
+            sql_gen_kwargs=self.sql_gen_kwargs,
+            default_catalog=self.default_catalog,
+            execute_log_level=level,
+            **self._extra_config,
+        )
+
+        adapter._connection_pool = self._connection_pool
+
+        return adapter
+
     @property
     def cursor(self) -> t.Any:
         return self._connection_pool.get_cursor()
@@ -1386,13 +1402,20 @@ class EngineAdapter:
 
         with self.transaction():
             for e in ensure_list(expressions):
-                sql = (
+                sql = t.cast(
+                    str,
                     self._to_sql(e, quote=quote_identifiers, **to_sql_kwargs)
                     if isinstance(e, exp.Expression)
-                    else e
+                    else e,
                 )
-                logger.debug(f"Executing SQL:\n{sql}")
-                self.cursor.execute(sql, **kwargs)
+                self._log_sql(sql)
+                self._execute(sql, **kwargs)
+
+    def _log_sql(self, sql: str) -> None:
+        logger.log(self._execute_log_level, "Executing SQL: %s", sql)
+
+    def _execute(self, sql: str, **kwargs: t.Any) -> None:
+        self.cursor.execute(sql, **kwargs)
 
     @contextlib.contextmanager
     def temp_table(
diff --git a/sqlmesh/core/engine_adapter/bigquery.py b/sqlmesh/core/engine_adapter/bigquery.py
index 83a8ee62..821954cb 100644
--- a/sqlmesh/core/engine_adapter/bigquery.py
+++ b/sqlmesh/core/engine_adapter/bigquery.py
@@ -5,8 +5,6 @@ import typing as t
 
 import pandas as pd
 from sqlglot import exp
-from sqlglot.errors import ErrorLevel
-from sqlglot.helper import ensure_list
 from sqlglot.transforms import remove_precision_parameterized_types
 
 from sqlmesh.core.dialect import to_schema
@@ -262,7 +260,7 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
         job_config = bigquery.job.LoadJobConfig(schema=self.__get_bq_schema(columns_to_types))
         if replace:
             job_config.write_disposition = bigquery.WriteDisposition.WRITE_TRUNCATE
-        logger.debug(f"Loading dataframe to BigQuery. Table Path: {table.path}")
+        logger.info(f"Loading dataframe to BigQuery. Table Path: {table.path}")
         # This client call does not support retry so we don't use the `_db_call` method.
         result = self.__retry(
             self.__db_load_table_from_dataframe,
@@ -496,58 +494,42 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
             **kwargs,
         )
 
-    def execute(
+    def _execute(
         self,
-        expressions: t.Union[str, exp.Expression, t.Sequence[exp.Expression]],
-        ignore_unsupported_errors: bool = False,
-        quote_identifiers: bool = True,
+        sql: str,
         **kwargs: t.Any,
     ) -> None:
         """Execute a sql query."""
         from google.cloud.bigquery import QueryJobConfig
         from google.cloud.bigquery.query import ConnectionProperty
 
-        to_sql_kwargs = (
-            {"unsupported_level": ErrorLevel.IGNORE} if ignore_unsupported_errors else {}
+        # BigQuery's Python DB API implementation does not support retries, so we have to implement them ourselves.
+        # So we update the cursor's query job and query data with the results of the new query job. This makes sure
+        # that other cursor based operations execute correctly.
+        session_id = self._session_id
+        connection_properties = (
+            [
+                ConnectionProperty(key="session_id", value=session_id),
+            ]
+            if session_id
+            else []
         )
 
-        for e in ensure_list(expressions):
-            sql = (
-                self._to_sql(e, quote=quote_identifiers, **to_sql_kwargs)
-                if isinstance(e, exp.Expression)
-                else e
-            )
-            logger.debug(f"Executing SQL:\n{sql}")
-
-            # BigQuery's Python DB API implementation does not support retries, so we have to implement them ourselves.
-            # So we update the cursor's query job and query data with the results of the new query job. This makes sure
-            # that other cursor based operations execute correctly.
-            session_id = self._session_id
-            connection_properties = (
-                [
-                    ConnectionProperty(key="session_id", value=session_id),
-                ]
-                if session_id
-                else []
-            )
-
-            job_config = QueryJobConfig(
-                **self._job_params, connection_properties=connection_properties
-            )
-            self._query_job = self._db_call(
-                self.client.query,
-                query=sql,
-                job_config=job_config,
-                timeout=self._extra_config.get("job_creation_timeout_seconds"),
-            )
-            results = self._db_call(
-                self._query_job.result,
-                timeout=self._extra_config.get("job_execution_timeout_seconds"),  # type: ignore
-            )
-            self._query_data = iter(results) if results.total_rows else iter([])
-            query_results = self._query_job._query_results
-            self.cursor._set_rowcount(query_results)
-            self.cursor._set_description(query_results.schema)
+        job_config = QueryJobConfig(**self._job_params, connection_properties=connection_properties)
+        self._query_job = self._db_call(
+            self.client.query,
+            query=sql,
+            job_config=job_config,
+            timeout=self._extra_config.get("job_creation_timeout_seconds"),
+        )
+        results = self._db_call(
+            self._query_job.result,
+            timeout=self._extra_config.get("job_execution_timeout_seconds"),  # type: ignore
+        )
+        self._query_data = iter(results) if results.total_rows else iter([])
+        query_results = self._query_job._query_results
+        self.cursor._set_rowcount(query_results)
+        self.cursor._set_description(query_results.schema)
 
     def _get_data_objects(self, schema_name: SchemaName) -> t.List[DataObject]:
         """
@@ -637,9 +619,7 @@ class _ErrorCounter:
             return False
         self.error_count += 1
         if self._is_retryable(error) and self.error_count <= self.num_retries:
-            logger.debug(
-                f"Retry Num {self.error_count} of {self.num_retries}. Error: {repr(error)}"
-            )
+            logger.info(f"Retry Num {self.error_count} of {self.num_retries}. Error: {repr(error)}")
             return True
         return False
 
diff --git a/sqlmesh/core/engine_adapter/databricks.py b/sqlmesh/core/engine_adapter/databricks.py
index 2db4451d..987eae7c 100644
--- a/sqlmesh/core/engine_adapter/databricks.py
+++ b/sqlmesh/core/engine_adapter/databricks.py
@@ -114,12 +114,13 @@ class DatabricksEngineAdapter(SparkEngineAdapter):
         if self.is_spark_session_cursor:
             return super()._fetch_native_df(query, quote_identifiers=quote_identifiers)
         if self._use_spark_session:
-            logger.debug(f"Executing SQL:\n{query}")
-            return self.spark.sql(
+            sql = (
                 self._to_sql(query, quote=quote_identifiers)
                 if isinstance(query, exp.Expression)
                 else query
             )
+            self._log_sql(sql)
+            return self.spark.sql(sql)
         self.execute(query)
         return self.cursor.fetchall_arrow().to_pandas()
 
diff --git a/sqlmesh/core/loader.py b/sqlmesh/core/loader.py
index 08409e88..b249f88e 100644
--- a/sqlmesh/core/loader.py
+++ b/sqlmesh/core/loader.py
@@ -118,9 +118,9 @@ class Loader(abc.ABC):
                 self._track_file(config_file)
                 config_mtimes[context_path].append(self._path_mtimes[config_file])
 
-        for config_file in context.sqlmesh_path.glob("config.*"):
+        for config_file in c.SQLMESH_PATH.glob("config.*"):
             self._track_file(config_file)
-            config_mtimes[context.sqlmesh_path].append(self._path_mtimes[config_file])
+            config_mtimes[c.SQLMESH_PATH].append(self._path_mtimes[config_file])
 
         self._config_mtimes = {path: max(mtimes) for path, mtimes in config_mtimes.items()}
 
@@ -439,7 +439,7 @@ class SqlMeshLoader(Loader):
                 self._loader._path_mtimes[model_path],
                 self._loader._macros_max_mtime,
                 self._loader._config_mtimes.get(self._context_path),
-                self._loader._config_mtimes.get(self._loader._context.sqlmesh_path),
+                self._loader._config_mtimes.get(c.SQLMESH_PATH),
             ]
             return "__".join(
                 [
diff --git a/sqlmesh/core/model/definition.py b/sqlmesh/core/model/definition.py
index 9aaa305b..a3adcf4b 100644
--- a/sqlmesh/core/model/definition.py
+++ b/sqlmesh/core/model/definition.py
@@ -8,7 +8,6 @@ import types
 import typing as t
 from difflib import unified_diff
 from pathlib import Path
-from textwrap import indent
 
 import pandas as pd
 from astor import to_source
@@ -987,10 +986,6 @@ class SqlModel(_SqlBasedModel):
             engine_adapter=engine_adapter,
             **kwargs,
         )
-        if query and engine_adapter and logger.isEnabledFor(logging.DEBUG):
-            logger.debug(
-                f"Rendered query for '{self.name}':\n{indent(query.sql(dialect=self.dialect, pretty=True), '  ')}"
-            )
         return query
 
     def render_definition(self, include_python: bool = True) -> t.List[exp.Expression]:
diff --git a/sqlmesh/core/model/kind.py b/sqlmesh/core/model/kind.py
index 1ece3e68..a68c5435 100644
--- a/sqlmesh/core/model/kind.py
+++ b/sqlmesh/core/model/kind.py
@@ -112,6 +112,9 @@ class ModelKindName(str, ModelKindMixin, Enum):
     def model_kind_name(self) -> t.Optional[ModelKindName]:
         return self
 
+    def __repr__(self) -> str:
+        return str(self)
+
 
 _unique_key_validator = field_validator("unique_key", mode="before")(parse_expressions)
 
diff --git a/sqlmesh/core/node.py b/sqlmesh/core/node.py
index eb10717c..890e28cd 100644
--- a/sqlmesh/core/node.py
+++ b/sqlmesh/core/node.py
@@ -361,6 +361,9 @@ class NodeType(str, Enum):
     MODEL = "model"
     AUDIT = "audit"
 
+    def __repr__(self) -> str:
+        return str(self)
+
 
 def str_or_exp_to_str(v: t.Any) -> t.Optional[str]:
     if isinstance(v, exp.Expression):
diff --git a/sqlmesh/core/plan/definition.py b/sqlmesh/core/plan/definition.py
index 936669ec..792fb53b 100644
--- a/sqlmesh/core/plan/definition.py
+++ b/sqlmesh/core/plan/definition.py
@@ -291,7 +291,6 @@ class Plan:
             if snapshot.change_category:
                 loaded_snapshots.append(LoadedSnapshotIntervals.from_snapshot(snapshot))
             else:
-                logger.debug(f"Got an unloaded snapshot. Snapshot: {snapshot.name}")
                 unloaded_snapshots.append(snapshot)
             for downstream_indirect in self.indirectly_modified.get(snapshot.name, set()):
                 downstream_snapshot = self._snapshot_mapping[downstream_indirect]
@@ -303,9 +302,6 @@ class Plan:
                         LoadedSnapshotIntervals.from_snapshot(downstream_snapshot)
                     )
                 else:
-                    logger.debug(
-                        f"Got an unloaded indirectly-modified snapshot. Snapshot: {downstream_snapshot.name}"
-                    )
                     unloaded_snapshots.append(downstream_snapshot)
         return loaded_snapshots, unloaded_snapshots
 
@@ -494,7 +490,6 @@ class Plan:
     def _add_restatements(self) -> None:
         def is_restateable_snapshot(snapshot: Snapshot) -> bool:
             if not self.is_dev and snapshot.disable_restatement:
-                logger.debug("Restatement is disabled for model '%s'.", snapshot.name)
                 return False
             return not snapshot.is_symbolic and not snapshot.is_seed
 
diff --git a/sqlmesh/core/snapshot/evaluator.py b/sqlmesh/core/snapshot/evaluator.py
index 58595878..3706a05b 100644
--- a/sqlmesh/core/snapshot/evaluator.py
+++ b/sqlmesh/core/snapshot/evaluator.py
@@ -319,8 +319,6 @@ class SnapshotEvaluator:
                 f"Cannot audit '{snapshot.name}' because it has not been versioned yet. Apply a plan first."
             )
 
-        logger.info("Auditing snapshot %s", snapshot.snapshot_id)
-
         if wap_id is not None:
             deployability_index = deployability_index or DeployabilityIndex.all_deployable()
             original_table_name = snapshot.table_name(
@@ -337,6 +335,12 @@ class SnapshotEvaluator:
             kwargs["this_model"] = exp.to_table(wap_table_name)
 
         results = []
+
+        audits_with_args = snapshot.audits_with_args
+
+        if audits_with_args:
+            logger.info("Auditing snapshot %s", snapshot.snapshot_id)
+
         for audit, audit_args in snapshot.audits_with_args:
             results.append(
                 self._audit(
@@ -702,7 +706,8 @@ class SnapshotEvaluator:
             **kwargs,
         )
         count, *_ = self.adapter.fetchone(
-            select("COUNT(*)").from_(query.subquery("audit")), quote_identifiers=True
+            select("COUNT(*)").from_(query.subquery("audit")),
+            quote_identifiers=True,
         )
         if count and raise_exception:
             audit_error = AuditError(
@@ -1113,10 +1118,20 @@ class IncrementalUnmanagedStrategy(MaterializableStrategy):
         model = snapshot.model
         if isinstance(model.kind, IncrementalUnmanagedKind) and model.kind.insert_overwrite:
             self.adapter.insert_overwrite_by_partition(
-                name, query_or_df, model.partitioned_by, columns_to_types=model.columns_to_types
+                name,
+                query_or_df,
+                model.partitioned_by,
+                columns_to_types=model.columns_to_types,
             )
         else:
-            self.append(snapshot, name, query_or_df, snapshots, deployability_index, **kwargs)
+            self.append(
+                snapshot,
+                name,
+                query_or_df,
+                snapshots,
+                deployability_index,
+                **kwargs,
+            )
 
 
 class FullRefreshStrategy(MaterializableStrategy):
diff --git a/sqlmesh/engines/spark/app.py b/sqlmesh/engines/spark/app.py
index 425d1a1e..6d03770c 100644
--- a/sqlmesh/engines/spark/app.py
+++ b/sqlmesh/engines/spark/app.py
@@ -51,6 +51,7 @@ def main(
             lambda: spark_session_db.connection(spark),
             dialect,
             multithreaded=ddl_concurrent_tasks > 1,
+            execute_log_level=logging.INFO,
         ),
         ddl_concurrent_tasks=ddl_concurrent_tasks,
     )
diff --git a/sqlmesh/schedulers/airflow/operators/targets.py b/sqlmesh/schedulers/airflow/operators/targets.py
index 4ff05304..79a11148 100644
--- a/sqlmesh/schedulers/airflow/operators/targets.py
+++ b/sqlmesh/schedulers/airflow/operators/targets.py
@@ -1,4 +1,5 @@
 import abc
+import logging
 import typing as t
 
 from airflow.exceptions import AirflowSkipException
@@ -60,6 +61,7 @@ class BaseTarget(abc.ABC, t.Generic[CP]):
                 connection_factory,
                 dialect,
                 multithreaded=self.ddl_concurrent_tasks > 1,
+                execute_log_level=logging.INFO,
                 **kwargs,
             ),
             ddl_concurrent_tasks=self.ddl_concurrent_tasks,


Test patch:
diff --git a/tests/core/test_context.py b/tests/core/test_context.py
index 6fdd02c6..3fdc2663 100644
--- a/tests/core/test_context.py
+++ b/tests/core/test_context.py
@@ -15,6 +15,7 @@ from sqlmesh.core.config import (
     EnvironmentSuffixTarget,
     ModelDefaultsConfig,
     SnowflakeConnectionConfig,
+    load_configs,
 )
 from sqlmesh.core.context import Context
 from sqlmesh.core.dialect import parse, schema_
@@ -313,9 +314,8 @@ model_defaults:
             ConfigError,
             match="User and password must be provided if using default authentication",
         ):
-            context._load_configs("config", paths=[project_config.parent])
-        context.sqlmesh_path = home_path
-        loaded_configs = context._load_configs("config", paths=[project_config.parent])
+            load_configs("config", paths=project_config.parent)
+        loaded_configs = load_configs("config", paths=project_config.parent, sqlmesh_path=home_path)
         assert len(loaded_configs) == 1
         snowflake_connection = list(loaded_configs.values())[0].gateways["snowflake"].connection  # type: ignore
         assert isinstance(snowflake_connection, SnowflakeConnectionConfig)