refactors and improvements

Author: themisvaltinos

Makefile

@@ -64,7 +64,7 @@ engine-up: engine-clickhouse-up engine-mssql-up engine-mysql-up engine-postgres-
 engine-down: engine-clickhouse-down engine-mssql-down engine-mysql-down engine-postgres-down engine-spark-down engine-trino-down
 
 fast-test:
-	pytest -n auto -m "fast and not cicdonly" && pytest -m "isolated" && pytest -m "isolated2"
+	pytest -n auto -m "fast and not cicdonly" && pytest -m "isolated" && pytest -m "registry_isolation"
 
 slow-test:
 	pytest -n auto -m "(fast or slow) and not cicdonly" && pytest -m "isolated"

pytest.ini

@@ -8,7 +8,7 @@ markers =
     remote: test that involves interacting with a remote DB
     cicdonly: test that only runs on CI/CD
     isolated: tests that need to run sequentially usually because they use fork
-    isolated2: tests that need to run isolated because they interfere
+    registry_isolation: tests that need to run isolated because they manipulate global registries or state
 
     # Test Domain Markers
     # default: core functionality

sqlmesh/core/loader.py

@@ -19,6 +19,7 @@
 
 from sqlmesh.core import constants as c
 from sqlmesh.core.audit import Audit, ModelAudit, StandaloneAudit, load_multiple_audits
+from sqlmesh.core.console import Console
 from sqlmesh.core.dialect import parse
 from sqlmesh.core.environment import EnvironmentStatements
 from sqlmesh.core.linter.rule import Rule
@@ -85,25 +86,32 @@ def get(self, path: Path) -> t.List[Model]:
 def _init_model_defaults(
     config: Config,
     selected_gateway: t.Optional[str],
-    defaults: t.Optional[t.Dict[str, t.Any]] = None,
+    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
-    _defaults = defaults
+    _defaults = model_loading_defaults
     _cache = cache
     _config = config
     _selected_gateway = selected_gateway
 
+    # Set the console passed from the parent process
+    if console is not None:
+        from sqlmesh.core.console import set_console
 
-def load_sql_models(path: Path) -> t.Tuple[Path, list[Model]]:
+        set_console(console)
+
+
+def load_sql_models(path: Path) -> t.List[Model]:
     assert _defaults
     assert _cache
 
     with open(path, "r", encoding="utf-8") as file:
         expressions = parse(file.read(), default_dialect=_defaults["dialect"])
     models = load_sql_based_models(expressions, path=Path(path).absolute(), **_defaults)
 
-    return (path, [] if _cache.put(models, path) else models)
+    return [] if _cache.put(models, path) else models
 
 
 def get_variables(gateway_name: t.Optional[str] = None) -> t.Dict[str, t.Any]:
@@ -528,7 +536,7 @@ def _load_sql_models(
                         models[model.fqn] = model
 
         if paths:
-            defaults = dict(
+            model_loading_defaults = dict(
                 get_variables=get_variables,
                 defaults=self.config.model_defaults.dict(),
                 macros=macros,
@@ -549,14 +557,14 @@ def _load_sql_models(
             errors: t.List[str] = []
             with create_process_pool_executor(
                 initializer=_init_model_defaults,
-                initargs=(self.config, gateway, defaults, cache),
+                initargs=(self.config, 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}
                 for future in concurrent.futures.as_completed(futures_to_paths):
                     path = futures_to_paths[future]
                     try:
-                        _, loaded = future.result()
+                        loaded = future.result()
                         for model in loaded or cache.get(path):
                             if model.fqn in models:
                                 errors.append(
@@ -572,7 +580,7 @@ def _load_sql_models(
 
             if errors:
                 error_string = "\n".join(errors)
-                raise ConfigError(f"Failed to load models\n\n{error_string}")
+                raise ConfigError(error_string)
 
         return models
 

sqlmesh/utils/process.py

@@ -24,8 +24,17 @@ def __enter__(self):
         return self
 
     def __exit__(self, *args):
+        self.shutdown(wait=True)
         return True
 
+    def shutdown(self, wait=True, cancel_futures=False):
+        """No-op method to match ProcessPoolExecutor API.
+
+        Since this executor runs synchronously, there are no background processes
+        or resources to shut down and all futures will have completed already.
+        """
+        pass
+
     def submit(self, fn, *args, **kwargs):
         """Execute the function synchronously and return a Future with the result."""
         future = Future()

tests/conftest.py

@@ -183,7 +183,7 @@ def validate(
 
 
 def pytest_collection_modifyitems(items, *args, **kwargs):
-    test_type_markers = {"fast", "slow", "docker", "remote", "isolated", "isolated2"}
+    test_type_markers = {"fast", "slow", "docker", "remote", "isolated", "registry_isolation"}
     for item in items:
         for marker in item.iter_markers():
             if marker.name in test_type_markers:

tests/core/test_loader.py

@@ -115,7 +115,7 @@ def duplicate_model_path(fpath):
         Context(paths=tmp_path, config=config)
 
 
-@pytest.mark.isolated2
+@pytest.mark.registry_isolation
 def test_duplicate_python_model_names_raise_error(tmp_path: Path) -> None:
     """Test python models with duplicate model names raises ConfigError if the functions are not identical."""
     init_example_project(tmp_path, dialect="duckdb")