address comments

Author: themisvaltinos

sqlmesh/core/loader.py

@@ -46,7 +46,6 @@
 
 
 if t.TYPE_CHECKING:
-    from sqlmesh.core.config import Config
     from sqlmesh.core.context import GenericContext
 
 
@@ -104,21 +103,21 @@ def get(self, path: Path) -> t.List[Model]:
 
 _defaults: t.Optional[t.Dict[str, t.Any]] = None
 _cache: t.Optional[CacheBase] = None
-_config: t.Optional[Config] = None
+_config_essentials: t.Optional[t.Dict[str, t.Any]] = None
 _selected_gateway: t.Optional[str] = None
 
 
 def _init_model_defaults(
-    config: Config,
+    config_essentials: t.Dict[str, t.Any],
     selected_gateway: t.Optional[str],
     model_loading_defaults: t.Optional[t.Dict[str, t.Any]] = None,
     cache: t.Optional[CacheBase] = None,
     console: t.Optional[Console] = None,
 ) -> None:
-    global _defaults, _cache, _config, _selected_gateway
+    global _defaults, _cache, _config_essentials, _selected_gateway
     _defaults = model_loading_defaults
     _cache = cache
-    _config = config
+    _config_essentials = config_essentials
     _selected_gateway = selected_gateway
 
     # Set the console passed from the parent process
@@ -140,22 +139,22 @@ def load_sql_models(path: Path) -> t.List[Model]:
 
 
 def get_variables(gateway_name: t.Optional[str] = None) -> t.Dict[str, t.Any]:
-    assert _config
+    assert _config_essentials
 
     gateway_name = gateway_name or _selected_gateway
 
     try:
-        gateway = _config.get_gateway(gateway_name)
+        gateway = _config_essentials["gateways"].get(gateway_name)
     except ConfigError:
         from sqlmesh.core.console import get_console
 
         get_console().log_warning(
-            f"Gateway '{gateway_name}' not found in project '{_config.project}'."
+            f"Gateway '{gateway_name}' not found in project '{_config_essentials['project']}'."
         )
         gateway = None
 
     return {
-        **_config.variables,
+        **_config_essentials["variables"],
         **(gateway.variables if gateway else {}),
         c.GATEWAY: gateway_name,
     }
@@ -174,7 +173,12 @@ def __init__(self, context: GenericContext, path: Path) -> None:
         self._variables_by_gateway: t.Dict[str, t.Dict[str, t.Any]] = {}
         self._console = get_console()
 
-        _init_model_defaults(self.config, self.context.selected_gateway)
+        self.config_essentials = {
+            "project": self.config.project,
+            "variables": self.config.variables,
+            "gateways": self.config.gateways,
+        }
+        _init_model_defaults(self.config_essentials, self.context.selected_gateway)
 
     def load(self) -> LoadedProject:
         """
@@ -539,6 +543,7 @@ def _load_sql_models(
         """Loads the sql models into a Dict"""
         models: UniqueKeyDict[str, Model] = UniqueKeyDict("models")
         paths: t.Set[Path] = set()
+        cached_paths: UniqueKeyDict[Path, t.List[Model]] = UniqueKeyDict("cached_paths")
 
         for path in self._glob_paths(
             self.config_path / c.MODELS,
@@ -550,14 +555,14 @@ def _load_sql_models(
 
             self._track_file(path)
             paths.add(path)
+            if cached_models := cache.get(path):
+                cached_paths[path] = cached_models
 
-        for path in paths.copy():
-            cached_models = cache.get(path)
-            if cached_models:
-                paths.remove(path)
-                for model in cached_models:
-                    if model.enabled:
-                        models[model.fqn] = model
+        for path, cached_models in cached_paths.items():
+            paths.remove(path)
+            for model in cached_models:
+                if model.enabled:
+                    models[model.fqn] = model
 
         if paths:
             model_loading_defaults = dict(
@@ -578,10 +583,18 @@ def _load_sql_models(
                 default_catalog_per_gateway=self.context.default_catalog_per_gateway,
             )
 
-            errors: t.List[str] = []
+            # if not c.MAX_FORK_WORKERS:
+            #     breakpoint()
+
             with create_process_pool_executor(
                 initializer=_init_model_defaults,
-                initargs=(self.config, gateway, model_loading_defaults, cache, self._console),
+                initargs=(
+                    self.config_essentials,
+                    gateway,
+                    model_loading_defaults,
+                    cache,
+                    self._console,
+                ),
                 max_workers=c.MAX_FORK_WORKERS,
             ) as pool:
                 futures_to_paths = {pool.submit(load_sql_models, path): path for path in paths}
@@ -591,7 +604,7 @@ def _load_sql_models(
                         loaded = future.result()
                         for model in loaded or cache.get(path):
                             if model.fqn in models:
-                                errors.append(
+                                raise ConfigError(
                                     self._failed_to_load_model_error(
                                         path, f"Duplicate SQL model name: '{model.name}'."
                                     )
@@ -600,11 +613,7 @@ def _load_sql_models(
                                 model._path = path
                                 models[model.fqn] = model
                     except Exception as ex:
-                        errors.append(self._failed_to_load_model_error(path, str(ex)))
-
-            if errors:
-                error_string = "\n".join(errors)
-                raise ConfigError(error_string)
+                        raise ConfigError(self._failed_to_load_model_error(path, ex))
 
         return models
 

sqlmesh/core/model/cache.py

@@ -65,7 +65,7 @@ def get_or_load(
         return models
 
     def put(self, models: t.List[Model], name: str, entry_id: str = "") -> bool:
-        if isinstance(models, list) and isinstance(seq_get(models, 0), (SqlModel, ExternalModel)):
+        if models and isinstance(seq_get(models, 0), (SqlModel, ExternalModel)):
             # make sure we preload full_depends_on
             for model in models:
                 model.full_depends_on

sqlmesh/core/model/schema.py

@@ -81,11 +81,10 @@ def process_models(completed_model: t.Optional[Model] = None) -> None:
                     )
                 )
 
-    errors: t.List[str] = []
     with optimized_query_cache_pool(optimized_query_cache) as executor:
         process_models()
 
-        while futures and not errors:
+        while futures:
             for future in as_completed(futures):
                 try:
                     futures.remove(future)
@@ -99,8 +98,4 @@ def process_models(completed_model: t.Optional[Model] = None) -> None:
                     _update_schema_with_model(schema, model)
                     process_models(completed_model=model)
                 except Exception as ex:
-                    errors.append(f"{ex}")
-
-    if errors:
-        error_string = "\n".join(errors)
-        raise SchemaError(f"Failed to update model schemas\n\n{error_string}")
+                    raise SchemaError(f"Failed to update model schemas\n\n{ex}")

sqlmesh/utils/process.py

@@ -3,7 +3,6 @@
 from concurrent.futures import Future, ProcessPoolExecutor
 import typing as t
 import multiprocessing as mp
-from sqlmesh.core import constants as c
 from sqlmesh.utils.windows import IS_WINDOWS
 
 
@@ -26,7 +25,7 @@ def __enter__(self):
 
     def __exit__(self, *args):
         self.shutdown(wait=True)
-        return True
+        return False
 
     def shutdown(self, wait=True, cancel_futures=False):
         """No-op method to match ProcessPoolExecutor API.
@@ -59,7 +58,7 @@ def map(self, fn, *iterables, timeout=None, chunksize=1):
 
 
 def create_process_pool_executor(
-    initializer: t.Callable, initargs: t.Tuple, max_workers: t.Optional[int] = c.MAX_FORK_WORKERS
+    initializer: t.Callable, initargs: t.Tuple, max_workers: t.Optional[int]
 ) -> PoolExecutor:
     if max_workers == 1 or IS_WINDOWS:
         return SynchronousPoolExecutor(

tests/core/test_integration.py

@@ -4133,12 +4133,12 @@ def test_plan_repairs_unrenderable_snapshot_state(
         f"name = '{target_snapshot.name}' AND identifier = '{target_snapshot.identifier}'",
     )
 
-    context.clear_caches()
-
-    target_snapshot_in_state = context.state_sync.get_snapshots([target_snapshot.snapshot_id])[
-        target_snapshot.snapshot_id
-    ]
     with pytest.raises(Exception):
+        context_copy = context.copy()
+        context_copy.clear_caches()
+        target_snapshot_in_state = context_copy.state_sync.get_snapshots(
+            [target_snapshot.snapshot_id]
+        )[target_snapshot.snapshot_id]
         target_snapshot_in_state.model.render_query_or_raise()
 
     # Repair the snapshot by creating a new version of it