Repository: TobikoData/sqlmesh
Pull request #3073 (2024-08-30 18:13:29)

feat: format check
Adds a `--check` flag to `sqlmesh format`

CLAassistant: [![CLA assistant check](https://cla-assistant.io/pull/badge/not_signed)](https://cla-assistant.io/TobikoData/sqlmesh?pullRequest=3073) <br/>Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our [Contributor License Agreement](https://cla-assistant.io/TobikoData/sqlmesh?pullRequest=3073) before we can accept your contribution.<br/><sub>You have signed the CLA already but the status is still pending? Let us [recheck](https://cla-assistant.io/check/TobikoData/sqlmesh?pullRequest=3073) it.</sub>
etonlels: @izeigerman I noticed that the `format` method is also being referenced [here](https://github.com/TobikoData/sqlmesh/blob/ae6f97ff3d3cd74e43283570cb8d6e23e34ac399/sqlmesh/magics.py#L692-L742), but I'm not familiar with what that's for exactly. Should a change be made there as well?
georgesittas: Hey 👋

> @izeigerman I noticed that the `format` method is also being referenced [here](https://github.com/TobikoData/sqlmesh/blob/ae6f97ff3d3cd74e43283570cb8d6e23e34ac399/sqlmesh/magics.py#L692-L742), but I'm not familiar with what that's for exactly. Should a change be made there as well?
 
Yeah, that's the notebook interface. We generally try to keep the CLI and notebook APIs in sync, so it makes sense to reflect your changes in there as well.
georgesittas: Let's update the documentation as well while we're at it (note that `format` is currently missing from notebook docs, but we should add that):

- https://sqlmesh.readthedocs.io/en/stable/reference/cli/
- https://sqlmesh.readthedocs.io/en/stable/reference/notebook/

The way I usually update these pages is I copy-paste the `--help` version of the command of interest in the corresponding markdown file.
etonlels: @georgesittas I updated the CLI docs, but when I go to view the docs for `%format` in a Jupyter notebook by pressing `Shift+Tab`, I get the docs for Python's builtin `format` function, i.e. the following:
```
Signature: format(value, format_spec='', /)
Docstring:
Return value.__format__(format_spec)

format_spec defaults to the empty string.
See the Format Specification Mini-Language section of help('FORMATTING') for
details.
Type:      builtin_function_or_method
```
I'm guessing that's why it's not currently documented. I'm not an expert with Jupyter notebooks (I never use them). Does it make sense to override this behavior? Is that even possible?
georgesittas: Interesting. In any case, we can also just fill in the docs by hand, mimicking the other magic command descriptions. Can also take this to the finish line myself later, so no worries.

@izeigerman do you want to take another look?
etonlels: @georgesittas @izeigerman I think I've addressed everything so far. PTAL when you get the chance. Thanks!

Patch:
diff --git a/Makefile b/Makefile
index 8ef4d722..e0e7390b 100644
--- a/Makefile
+++ b/Makefile
@@ -91,13 +91,13 @@ 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 -n auto -m "fast and not cicdonly" && pytest -m "isolated"
 
 slow-test:
-	pytest -n auto -m "(fast or slow) and not cicdonly"
+	pytest -n auto -m "(fast or slow) and not cicdonly" && pytest -m "isolated"
 
 cicd-test:
-	pytest -n auto -m "fast or slow" --junitxml=test-results/junit-cicd.xml
+	pytest -n auto -m "fast or slow" --junitxml=test-results/junit-cicd.xml && pytest -m "isolated"
 
 core-fast-test:
 	pytest -n auto -m "fast and not web and not github and not dbt and not airflow and not jupyter"
@@ -167,45 +167,45 @@ engine-%-down:
 ##################
 
 clickhouse-test: engine-clickhouse-up
-	pytest -n auto -x -m "clickhouse" --junitxml=test-results/junit-clickhouse.xml
+	pytest -n auto -x -m "clickhouse" --retries 3 --junitxml=test-results/junit-clickhouse.xml
 
 clickhouse-cluster-test: engine-clickhouse-up
-	pytest -n auto -x -m "clickhouse_cluster" --junitxml=test-results/junit-clickhouse-cluster.xml
+	pytest -n auto -x -m "clickhouse_cluster" --retries 3 --junitxml=test-results/junit-clickhouse-cluster.xml
 
 duckdb-test: engine-duckdb-install
-	pytest -n auto -x -m "duckdb" --junitxml=test-results/junit-duckdb.xml
+	pytest -n auto -x -m "duckdb" --retries 3 --junitxml=test-results/junit-duckdb.xml
 
 mssql-test: engine-mssql-up
-	pytest -n auto -x -m "mssql" --junitxml=test-results/junit-mssql.xml
+	pytest -n auto -x -m "mssql" --retries 3 --junitxml=test-results/junit-mssql.xml
 
 mysql-test: engine-mysql-up
-	pytest -n auto -x -m "mysql" --junitxml=test-results/junit-mysql.xml
+	pytest -n auto -x -m "mysql" --retries 3 --junitxml=test-results/junit-mysql.xml
 
 postgres-test: engine-postgres-up
-	pytest -n auto -x -m "postgres" --junitxml=test-results/junit-postgres.xml
+	pytest -n auto -x -m "postgres" --retries 3 --junitxml=test-results/junit-postgres.xml
 
 spark-test: engine-spark-up
-	pytest -n auto -x -m "spark or pyspark" --junitxml=test-results/junit-spark.xml
+	pytest -n auto -x -m "spark or pyspark" --retries 3 --junitxml=test-results/junit-spark.xml
 
 trino-test: engine-trino-up
-	pytest -n auto -x -m "trino or trino_iceberg or trino_delta" --retries 2 --junitxml=test-results/junit-trino.xml
+	pytest -n auto -x -m "trino or trino_iceberg or trino_delta" --retries 3 --junitxml=test-results/junit-trino.xml
 
 #################
 # Cloud Engines #
 #################
 
 snowflake-test: guard-SNOWFLAKE_ACCOUNT guard-SNOWFLAKE_WAREHOUSE guard-SNOWFLAKE_DATABASE guard-SNOWFLAKE_USER guard-SNOWFLAKE_PASSWORD engine-snowflake-install
-	pytest -n auto -x -m "snowflake" --junitxml=test-results/junit-snowflake.xml
+	pytest -n auto -x -m "snowflake" --retries 3 --junitxml=test-results/junit-snowflake.xml
 
 bigquery-test: guard-BIGQUERY_KEYFILE engine-bigquery-install
-	pytest -n auto -x -m "bigquery" --junitxml=test-results/junit-bigquery.xml
+	pytest -n auto -x -m "bigquery" --retries 3 --junitxml=test-results/junit-bigquery.xml
 
 databricks-test: guard-DATABRICKS_CATALOG guard-DATABRICKS_SERVER_HOSTNAME guard-DATABRICKS_HTTP_PATH guard-DATABRICKS_ACCESS_TOKEN guard-DATABRICKS_CONNECT_VERSION engine-databricks-install
 	pip install 'databricks-connect==${DATABRICKS_CONNECT_VERSION}'
-	pytest -n auto -x -m "databricks" --junitxml=test-results/junit-databricks.xml
+	pytest -n auto -x -m "databricks" --retries 3 --junitxml=test-results/junit-databricks.xml
 
 redshift-test: guard-REDSHIFT_HOST guard-REDSHIFT_USER guard-REDSHIFT_PASSWORD guard-REDSHIFT_DATABASE engine-redshift-install
-	pytest -n auto -x -m "redshift" --junitxml=test-results/junit-redshift.xml
+	pytest -n auto -x -m "redshift" --retries 3 --junitxml=test-results/junit-redshift.xml
 
 clickhouse-cloud-test: guard-CLICKHOUSE_CLOUD_HOST guard-CLICKHOUSE_CLOUD_USERNAME guard-CLICKHOUSE_CLOUD_PASSWORD engine-clickhouse-install
-	pytest -n auto -x -m "clickhouse_cloud" --junitxml=test-results/junit-clickhouse-cloud.xml
\ No newline at end of file
+	pytest -n auto -x -m "clickhouse_cloud" --retries 3 --junitxml=test-results/junit-clickhouse-cloud.xml
diff --git a/docs/reference/cli.md b/docs/reference/cli.md
index e1957f0e..d7a0cc05 100644
--- a/docs/reference/cli.md
+++ b/docs/reference/cli.md
@@ -184,6 +184,8 @@ Options:
                               trailing.
   --max-text-width INTEGER    The max number of characters in a segment before
                               creating new lines in pretty mode.
+  --check                     Whether or not to check formatting (but not
+                              actually format anything).
   --help                      Show this message and exit.
 ```
 
@@ -287,7 +289,8 @@ Options:
   --no-gaps                       Ensure that new snapshots have no data gaps
                                   when comparing to existing snapshots for
                                   matching models in the target environment.
-  --skip-backfill                 Skip the backfill step.
+  --skip-backfill, --dry-run      Skip the backfill step and only create a
+                                  virtual update for the plan.
   --forward-only                  Create a plan for forward-only changes.
   --allow-destructive-model TEXT  Allow destructive forward-only changes to
                                   models whose names match the expression.
diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md
index 7c93d254..4734b39d 100644
--- a/docs/reference/configuration.md
+++ b/docs/reference/configuration.md
@@ -16,10 +16,10 @@ This section describes the other root level configuration parameters.
 
 Configuration options for SQLMesh project directories.
 
-| Option            | Description                                                                                                        |     Type     | Required |
-| ----------------- | ------------------------------------------------------------------------------------------------------------------ | :----------: | :------: |
-| `ignore_patterns` | Files that match glob patterns specified in this list are ignored when scanning the project folder (Default: `[]`) | list[string] |    N     |
-| `project`         | The project name of this config. Used for [multi-repo setups](../guides/multi_repo.md).                            | string       |    N     |
+| Option             | Description                                                                                                        |     Type     | Required |
+| ------------------ | ------------------------------------------------------------------------------------------------------------------ | :----------: | :------: |
+| `ignore_patterns`  | Files that match glob patterns specified in this list are ignored when scanning the project folder (Default: `[]`) | list[string] |    N     |
+| `project`          | The project name of this config. Used for [multi-repo setups](../guides/multi_repo.md).                            | string       |    N     |
 
 ### Environments
 
@@ -291,3 +291,8 @@ You can disable collection of anonymized usage information with these methods:
 
 - Set the root `disable_anonymized_analytics: true` key in your SQLMesh project configuration file
 - Execute SQLMesh commands with an environment variable `SQLMESH__DISABLE_ANONYMIZED_ANALYTICS` set to `1`, `true`, `t`, `yes`, or `y`
+
+## Parallel loading
+SQLMesh by default uses all of your cores when loading models and snapshots. It takes advantage of `fork` which is not available on Windows. The default is to use the same number of workers as cores on your machine if fork is available.
+
+You can override this setting by setting the environment variable `MAX_FORK_WORKERS`. A value of 1 will disable forking and load things sequentially.
diff --git a/docs/reference/notebook.md b/docs/reference/notebook.md
index 5b34da79..b000d1d5 100644
--- a/docs/reference/notebook.md
+++ b/docs/reference/notebook.md
@@ -95,7 +95,7 @@ options:
 %plan [--start START] [--end END] [--execution-time EXECUTION_TIME]
             [--create-from CREATE_FROM] [--skip-tests]
             [--restate-model [RESTATE_MODEL ...]] [--no-gaps]
-            [--skip-backfill] [--forward-only]
+            [--skip-backfill, --dry-run] [--forward-only]
             [--effective-from EFFECTIVE_FROM] [--no-prompts] [--auto-apply]
             [--no-auto-categorization] [--include-unmodified]
             [--select-model [SELECT_MODEL ...]]
@@ -127,7 +127,8 @@ options:
   --no-gaps, -g         Ensure that new snapshots have no data gaps when
                         comparing to existing snapshots for matching models in
                         the target environment.
-  --skip-backfill       Skip the backfill step.
+  --skip-backfill, --dry-run
+                        Skip the backfill step and only create a virtual update for the plan.
   --forward-only        Create a plan for forward-only changes.
   --effective-from EFFECTIVE_FROM
                         The effective date from which to apply forward-only
@@ -437,3 +438,32 @@ options:
   --read READ    The input dialect of the sql string.
   --write WRITE  The output dialect of the sql string.
 ```
+
+#### format
+```
+%format [--transpile TRANSPILE] [--append-newline] [--normalize]
+              [--pad PAD] [--indent INDENT]
+              [--normalize-functions NORMALIZE_FUNCTIONS] [--leading-comma]
+              [--max-text-width MAX_TEXT_WIDTH] [--check]
+
+Format all SQL models and audits.
+
+options:
+  --transpile TRANSPILE, -t TRANSPILE
+                        Transpile project models to the specified dialect.
+  --append-newline      Whether or not to append a newline to the end of the
+                        file.
+  --normalize           Whether or not to normalize identifiers to lowercase.
+  --pad PAD             Determines the pad size in a formatted string.
+  --indent INDENT       Determines the indentation size in a formatted string.
+  --normalize-functions NORMALIZE_FUNCTIONS
+                        Whether or not to normalize all function names.
+                        Possible values are: 'upper', 'lower'
+  --leading-comma       Determines whether or not the comma is leading or
+                        trailing in select expressions. Default is trailing.
+  --max-text-width MAX_TEXT_WIDTH
+                        The max number of characters in a segment before
+                        creating new lines in pretty mode.
+  --check               Whether or not to check formatting (but not actually
+                        format anything).
+```
diff --git a/sqlmesh/cli/main.py b/sqlmesh/cli/main.py
index 411e2a9e..29daf997 100644
--- a/sqlmesh/cli/main.py
+++ b/sqlmesh/cli/main.py
@@ -255,12 +255,19 @@ def evaluate(
     type=int,
     help="The max number of characters in a segment before creating new lines in pretty mode.",
 )
+@click.option(
+    "--check",
+    is_flag=True,
+    help="Whether or not to check formatting (but not actually format anything).",
+    default=None,
+)
 @click.pass_context
 @error_handler
 @cli_analytics
 def format(ctx: click.Context, **kwargs: t.Any) -> None:
     """Format all SQL models and audits."""
-    ctx.obj.format(**{k: v for k, v in kwargs.items() if v is not None})
+    if not ctx.obj.format(**{k: v for k, v in kwargs.items() if v is not None}):
+        ctx.exit(1)
 
 
 @cli.command("diff")
@@ -303,8 +310,9 @@ def diff(ctx: click.Context, environment: t.Optional[str] = None) -> None:
 )
 @click.option(
     "--skip-backfill",
+    "--dry-run",
     is_flag=True,
-    help="Skip the backfill step.",
+    help="Skip the backfill step and only create a virtual update for the plan.",
 )
 @click.option(
     "--forward-only",
diff --git a/sqlmesh/core/constants.py b/sqlmesh/core/constants.py
index 77cd59d6..cc04a4c4 100644
--- a/sqlmesh/core/constants.py
+++ b/sqlmesh/core/constants.py
@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import datetime
+import os
+import typing as t
 from pathlib import Path
 
 SQLMESH = "sqlmesh"
@@ -28,6 +30,21 @@ DEFAULT_TIME_COLUMN_FORMAT = "%Y-%m-%d"
 MAX_MODEL_DEFINITION_SIZE = 10000
 """Maximum number of characters in a model definition"""
 
+
+# The maximum number of fork processes, used for loading projects
+# None means default to process pool, 1 means don't fork, :N is number of processes
+# Factors in the number of available CPUs even if the process is bound to a subset of them
+# (e.g. via taskset) to avoid oversubscribing the system and causing kill signals
+if hasattr(os, "fork"):
+    try:
+        MAX_FORK_WORKERS: t.Optional[int] = int(os.getenv("MAX_FORK_WORKERS"))  # type: ignore
+    except TypeError:
+        MAX_FORK_WORKERS = (
+            len(os.sched_getaffinity(0)) if hasattr(os, "sched_getaffinity") else None
+        )
+else:
+    MAX_FORK_WORKERS = 1
+
 EPOCH = datetime.date(1970, 1, 1)
 
 DEFAULT_MAX_LIMIT = 1000
diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py
index a298c754..add5f099 100644
--- a/sqlmesh/core/context.py
+++ b/sqlmesh/core/context.py
@@ -292,10 +292,13 @@ class GenericContext(BaseContext, t.Generic[C]):
         load: Whether or not to automatically load all models and macros (default True).
         console: The rich instance used for printing out CLI command results.
         users: A list of users to make known to SQLMesh.
-        config_type: The type of config object to use (default Config).
     """
 
     CONFIG_TYPE: t.Type[C]
+    """The type of config object to use (default: Config)."""
+
+    PLAN_BUILDER_TYPE = PlanBuilder
+    """The type of plan builder object to use (default: PlanBuilder)."""
 
     def __init__(
         self,
@@ -860,17 +863,18 @@ class GenericContext(BaseContext, t.Generic[C]):
         self,
         transpile: t.Optional[str] = None,
         append_newline: t.Optional[bool] = None,
+        *,
+        check: t.Optional[bool] = None,
         **kwargs: t.Any,
-    ) -> None:
+    ) -> bool:
         """Format all SQL models and audits."""
         format_targets = {**self._models, **self._audits}
         for target in format_targets.values():
             if not target._path.suffix == ".sql":
                 continue
             with open(target._path, "r+", encoding="utf-8") as file:
-                expressions = parse(
-                    file.read(), default_dialect=self.config_for_node(target).dialect
-                )
+                before = file.read()
+                expressions = parse(before, default_dialect=self.config_for_node(target).dialect)
                 if transpile and is_meta_expression(expressions[0]):
                     for prop in expressions[0].expressions:
                         if prop.name.lower() == "dialect":
@@ -882,15 +886,18 @@ class GenericContext(BaseContext, t.Generic[C]):
                             )
                 format = self.config_for_node(target).format
                 opts = {**format.generator_options, **kwargs}
-                file.seek(0)
-                file.write(
-                    format_model_expressions(expressions, transpile or target.dialect, **opts)
-                )
+                after = format_model_expressions(expressions, transpile or target.dialect, **opts)
                 if append_newline is None:
                     append_newline = format.append_newline
                 if append_newline:
-                    file.write("\n")
-                file.truncate()
+                    after += "\n"
+                if not check:
+                    file.seek(0)
+                    file.write(after)
+                    file.truncate()
+                elif before != after:
+                    return False
+        return True
 
     @python_api_analytics
     def plan(
@@ -1163,7 +1170,7 @@ class GenericContext(BaseContext, t.Generic[C]):
                     days=1
                 )
 
-        return PlanBuilder(
+        return self.PLAN_BUILDER_TYPE(
             context_diff=context_diff,
             start=start,
             end=end,
diff --git a/sqlmesh/core/engine_adapter/bigquery.py b/sqlmesh/core/engine_adapter/bigquery.py
index d370c03a..93a9a761 100644
--- a/sqlmesh/core/engine_adapter/bigquery.py
+++ b/sqlmesh/core/engine_adapter/bigquery.py
@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import logging
 import typing as t
+from collections import defaultdict
 
 import pandas as pd
 from sqlglot import exp
@@ -51,8 +52,6 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
     MAX_TABLE_COMMENT_LENGTH = 1024
     MAX_COLUMN_COMMENT_LENGTH = 1024
 
-    # SQL is not supported for adding columns to structs: https://cloud.google.com/bigquery/docs/managing-table-schemas#api_1
-    # Can explore doing this with the API in the future
     SCHEMA_DIFFER = SchemaDiffer(
         compatible_types={
             exp.DataType.build("INT64", dialect=DIALECT): {
@@ -83,6 +82,8 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
                 exp.DataType.build("BYTES", dialect=DIALECT).this,
             },
         },
+        support_nested_operations=True,
+        support_nested_drop=False,
     )
 
     @property
@@ -224,6 +225,23 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
                 columns["_PARTITIONDATE"] = exp.DataType.build("DATE")
         return columns
 
+    def alter_table(
+        self,
+        alter_expressions: t.List[exp.Alter],
+    ) -> None:
+        """
+        Performs the alter statements to change the current table into the structure of the target table,
+        and uses the API to add columns to structs, where SQL is not supported.
+        """
+
+        nested_fields, non_nested_expressions = self._split_alter_expressions(alter_expressions)
+
+        if nested_fields:
+            self._update_table_schema_nested_fields(nested_fields, alter_expressions[0].this)
+
+        if non_nested_expressions:
+            super().alter_table(non_nested_expressions)
+
     def fetchone(
         self,
         query: t.Union[exp.Expression, str],
@@ -261,6 +279,64 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
         )
         return list(self._query_data)
 
+    def _split_alter_expressions(
+        self,
+        alter_expressions: t.List[exp.Alter],
+    ) -> t.Tuple[t.Dict[str, list[t.Tuple[str, str]]], t.List[exp.Alter]]:
+        """
+        Returns a dictionary of the nested fields to add and a list of the non-nested alter expressions.
+        """
+        nested_fields_to_add: t.Dict[str, list[t.Tuple[str, str]]] = defaultdict(list)
+        non_nested_expressions = []
+
+        for alter_expression in alter_expressions:
+            action = alter_expression.args["actions"][0]
+            if (
+                isinstance(action, exp.ColumnDef)
+                and isinstance(action.this, exp.Dot)
+                and isinstance(action.kind, exp.DataType)
+            ):
+                record = action.this.this.sql(dialect="bigquery")
+                field = action.this.expression.sql(dialect="bigquery")
+                data_type = action.kind.sql(dialect="bigquery")
+                nested_fields_to_add[record].append((data_type, field))
+            else:
+                non_nested_expressions.append(alter_expression)
+
+        return nested_fields_to_add, non_nested_expressions
+
+    def _update_table_schema_nested_fields(
+        self, nested_fields_to_add: t.Dict[str, list[t.Tuple[str, str]]], table_name: str
+    ) -> None:
+        """
+        Updates a BigQuery table schema by adding the new nested fields provided.
+        """
+        from google.cloud import bigquery
+
+        table = self._get_table(table_name)
+        original_schema = table.schema
+        new_schema = []
+
+        for field in original_schema:
+            current_fields = list(field.fields)
+            if field.name in nested_fields_to_add:
+                current_fields.extend(
+                    bigquery.SchemaField(new_field[1], new_field[0], mode="NULLABLE")
+                    for new_field in nested_fields_to_add[field.name]
+                )
+            if current_fields:
+                new_schema.append(
+                    bigquery.SchemaField(
+                        field.name, "RECORD", mode="NULLABLE", fields=tuple(current_fields)
+                    )
+                )
+            else:
+                new_schema.append(field)
+
+        if new_schema != original_schema:
+            table.schema = new_schema
+            self.client.update_table(table, ["schema"])
+
     def __load_pandas_to_table(
         self,
         table: bigquery.Table,
diff --git a/sqlmesh/core/engine_adapter/databricks.py b/sqlmesh/core/engine_adapter/databricks.py
index bd368ada..da77ff35 100644
--- a/sqlmesh/core/engine_adapter/databricks.py
+++ b/sqlmesh/core/engine_adapter/databricks.py
@@ -39,6 +39,7 @@ class DatabricksEngineAdapter(SparkEngineAdapter):
     SCHEMA_DIFFER = SchemaDiffer(
         support_positional_add=True,
         support_nested_operations=True,
+        support_nested_drop=True,
         array_element_selector="element",
         parameterized_type_defaults={
             exp.DataType.build("DECIMAL", dialect=DIALECT).this: [(10, 0), (0,)],
diff --git a/sqlmesh/core/loader.py b/sqlmesh/core/loader.py
index ee22cd6d..52c121a9 100644
--- a/sqlmesh/core/loader.py
+++ b/sqlmesh/core/loader.py
@@ -3,11 +3,10 @@ from __future__ import annotations
 import abc
 import linecache
 import logging
-import multiprocessing as mp
 import os
 import typing as t
 from collections import defaultdict
-from concurrent.futures import ProcessPoolExecutor, as_completed
+from concurrent.futures import as_completed
 from dataclasses import dataclass
 from pathlib import Path
 
@@ -25,10 +24,10 @@ from sqlmesh.core.model import (
     ModelCache,
     OptimizedQueryCache,
     SeedModel,
-    SqlModel,
     create_external_model,
     load_sql_based_model,
 )
+from sqlmesh.core.model.cache import optimized_query_cache_pool, load_optimized_query_cache
 from sqlmesh.core.model import model as model_registry
 from sqlmesh.utils import UniqueKeyDict
 from sqlmesh.utils.dag import DAG
@@ -549,7 +548,7 @@ def update_model_schemas(
     schema = MappingSchema(normalize=False)
     optimized_query_cache: OptimizedQueryCache = OptimizedQueryCache(context_path / c.CACHE)
 
-    if not hasattr(os, "fork") or "PYTEST_CURRENT_TEST" in os.environ:
+    if c.MAX_FORK_WORKERS == 1:
         _update_model_schemas_sequential(dag, models, schema, optimized_query_cache)
     else:
         _update_model_schemas_parallel(dag, models, schema, optimized_query_cache)
@@ -610,13 +609,9 @@ def _update_model_schemas_parallel(
                 del graph[name]
                 model = models[name]
                 model.update_schema(schema)
-                futures.add(executor.submit(_load_optimized_query_cache, model))
+                futures.add(executor.submit(load_optimized_query_cache, model))
 
-    with ProcessPoolExecutor(
-        mp_context=mp.get_context("fork"),
-        initializer=_init_optimized_query_cache,
-        initargs=(optimized_query_cache,),
-    ) as executor:
+    with optimized_query_cache_pool(optimized_query_cache) as executor:
         process_models()
 
         while futures:
@@ -629,20 +624,3 @@ def _update_model_schemas_parallel(
 
                 _update_schema_with_model(schema, model)
                 process_models(completed_model=model)
-
-
-_optimized_query_cache: t.Optional[OptimizedQueryCache] = None
-
-
-def _init_optimized_query_cache(optimized_query_cache: OptimizedQueryCache) -> None:
-    global _optimized_query_cache
-    _optimized_query_cache = optimized_query_cache
-
-
-def _load_optimized_query_cache(model: Model) -> t.Tuple[str, t.Optional[str]]:
-    assert _optimized_query_cache
-    if isinstance(model, SqlModel):
-        entry_name = _optimized_query_cache.put(model)
-    else:
-        entry_name = None
-    return model.fqn, entry_name
diff --git a/sqlmesh/core/model/cache.py b/sqlmesh/core/model/cache.py
index 62bf9e72..4a79634a 100644
--- a/sqlmesh/core/model/cache.py
+++ b/sqlmesh/core/model/cache.py
@@ -1,12 +1,15 @@
 from __future__ import annotations
 
 import logging
+import multiprocessing as mp
 import typing as t
+from concurrent.futures import ProcessPoolExecutor
 from pathlib import Path
 
 from sqlglot import exp
 from sqlglot.optimizer.simplify import gen
 
+from sqlmesh.core import constants as c
 from sqlmesh.core.model.definition import Model, SqlModel, _Model
 from sqlmesh.utils.cache import FileCache
 from sqlmesh.utils.hashing import crc32
@@ -15,6 +18,8 @@ from dataclasses import dataclass
 
 logger = logging.getLogger(__name__)
 
+T = t.TypeVar("T")
+
 
 class ModelCache:
     """File-based cache implementation for model definitions.
@@ -128,6 +133,49 @@ class OptimizedQueryCache:
         return f"{model.name}_{crc32(hash_data)}"
 
 
+def optimized_query_cache_pool(optimized_query_cache: OptimizedQueryCache) -> ProcessPoolExecutor:
+    return ProcessPoolExecutor(
+        mp_context=mp.get_context("fork"),
+        initializer=_init_optimized_query_cache,
+        initargs=(optimized_query_cache,),
+        max_workers=c.MAX_FORK_WORKERS,
+    )
+
+
+@t.overload
+def load_optimized_query_cache(
+    model_or_tuple: t.Tuple[Model, T],
+) -> t.Tuple[T, t.Optional[str]]: ...
+
+
+@t.overload
+def load_optimized_query_cache(model_or_tuple: Model) -> t.Tuple[str, t.Optional[str]]: ...
+
+
+def load_optimized_query_cache(model_or_tuple):  # type: ignore
+    assert _optimized_query_cache
+
+    if isinstance(model_or_tuple, _Model):
+        model = model_or_tuple
+        key = None
+    else:
+        model, key = model_or_tuple
+
+    if isinstance(model, SqlModel):
+        entry_name = _optimized_query_cache.put(model)
+    else:
+        entry_name = None
+    return key or model.fqn, entry_name
+
+
+_optimized_query_cache: t.Optional[OptimizedQueryCache] = None
+
+
+def _init_optimized_query_cache(optimized_query_cache: OptimizedQueryCache) -> None:
+    global _optimized_query_cache
+    _optimized_query_cache = optimized_query_cache
+
+
 def _mapping_schema_hash_data(schema: t.Dict[str, t.Any]) -> t.List[str]:
     keys = sorted(schema) if all(isinstance(v, dict) for v in schema.values()) else schema
 
diff --git a/sqlmesh/core/model/decorator.py b/sqlmesh/core/model/decorator.py
index 91327bf8..23eac1ad 100644
--- a/sqlmesh/core/model/decorator.py
+++ b/sqlmesh/core/model/decorator.py
@@ -134,6 +134,8 @@ class model(registry_decorator):
 
         if self.is_sql:
             query = MacroFunc(this=exp.Anonymous(this=entrypoint))
+            if self.columns:
+                common_kwargs["columns"] = self.columns
             return create_sql_model(
                 self.name, query, module_path=module_path, dialect=dialect, **common_kwargs
             )
diff --git a/sqlmesh/core/model/definition.py b/sqlmesh/core/model/definition.py
index 5ea0979e..96f5fbc9 100644
--- a/sqlmesh/core/model/definition.py
+++ b/sqlmesh/core/model/definition.py
@@ -961,7 +961,7 @@ class _Model(ModelMeta, frozen=True):
         if self._full_depends_on is None:
             depends_on = self.depends_on_ or set()
 
-            query = self.render_query(optimize=False)
+            query = self.render_query(needs_optimization=False)
             if query is not None:
                 depends_on |= d.find_tables(
                     query, default_catalog=self.default_catalog, dialect=self.dialect
diff --git a/sqlmesh/core/renderer.py b/sqlmesh/core/renderer.py
index c9180c20..c293f1e4 100644
--- a/sqlmesh/core/renderer.py
+++ b/sqlmesh/core/renderer.py
@@ -376,7 +376,7 @@ class QueryRenderer(BaseExpressionRenderer):
         table_mapping: t.Optional[t.Dict[str, str]] = None,
         deployability_index: t.Optional[DeployabilityIndex] = None,
         expand: t.Iterable[str] = tuple(),
-        optimize: bool = True,
+        needs_optimization: bool = True,
         runtime_stage: RuntimeStage = RuntimeStage.LOADING,
         **kwargs: t.Any,
     ) -> t.Optional[exp.Query]:
@@ -393,7 +393,8 @@ class QueryRenderer(BaseExpressionRenderer):
             expand: Expand referenced models as subqueries. This is used to bypass backfills when running queries
                 that depend on materialized tables.  Model definitions are inlined and can thus be run end to
                 end on the fly.
-            optimize: Whether to optimize the query.
+            needs_optimization: Whether or not an optimization should be attempted
+                (if passing False, it still may return a cached optimized query).
             runtime_stage: Indicates the current runtime stage, for example if we're still loading the project, etc.
             kwargs: Additional kwargs to pass to the renderer.
 
@@ -402,7 +403,7 @@ class QueryRenderer(BaseExpressionRenderer):
         """
 
         should_cache = self._should_cache(
-            runtime_stage, start, end, execution_time, not optimize, *kwargs.values()
+            runtime_stage, start, end, execution_time, *kwargs.values()
         )
 
         if should_cache and self._optimized_cache:
@@ -417,7 +418,7 @@ class QueryRenderer(BaseExpressionRenderer):
                     table_mapping=table_mapping,
                     deployability_index=deployability_index,
                     runtime_stage=runtime_stage,
-                    normalize_identifiers=optimize,
+                    normalize_identifiers=needs_optimization,
                     **kwargs,
                 )
             except ParsetimeAdapterCallError:
@@ -439,7 +440,7 @@ class QueryRenderer(BaseExpressionRenderer):
                 )
                 raise
 
-            if optimize:
+            if needs_optimization:
                 deps = d.find_tables(
                     query, default_catalog=self._default_catalog, dialect=self._dialect
                 )
@@ -449,7 +450,7 @@ class QueryRenderer(BaseExpressionRenderer):
                 if should_cache:
                     self._optimized_cache = query
 
-        if optimize:
+        if needs_optimization:
             query = self._resolve_tables(
                 query,
                 snapshots=snapshots,
diff --git a/sqlmesh/core/schema_diff.py b/sqlmesh/core/schema_diff.py
index 7b6123c7..a333b60a 100644
--- a/sqlmesh/core/schema_diff.py
+++ b/sqlmesh/core/schema_diff.py
@@ -301,6 +301,8 @@ class SchemaDiffer(PydanticModel):
             specific position in the set of existing columns.
         support_nested_operations: Whether the engine for which the diff is being computed supports modifications to
             nested data types like STRUCTs and ARRAYs.
+        support_nested_drop: Whether the engine for which the diff is being computed supports removing individual
+            columns of nested STRUCTs.
         compatible_types: Types that are compatible and automatically coerced in actions like UNION ALL. Dict key is data
             type, and value is the set of types that are compatible with it.
         support_coercing_compatible_types: Whether or not the engine for which the diff is being computed supports direct
@@ -321,6 +323,7 @@ class SchemaDiffer(PydanticModel):
 
     support_positional_add: bool = False
     support_nested_operations: bool = False
+    support_nested_drop: bool = False
     array_element_selector: str = ""
     compatible_types: t.Dict[exp.DataType, t.Set[exp.DataType]] = {}
     support_coercing_compatible_types: bool = False
@@ -451,6 +454,15 @@ class SchemaDiffer(PydanticModel):
         )
         return operations
 
+    def _requires_drop_alteration(
+        self, current_struct: exp.DataType, new_struct: exp.DataType
+    ) -> bool:
+        for current_pos, current_kwarg in enumerate(current_struct.expressions.copy()):
+            new_pos, _ = self._get_matching_kwarg(current_kwarg, new_struct, current_pos)
+            if new_pos is None:
+                return True
+        return False
+
     def _resolve_drop_operation(
         self,
         parent_columns: t.List[TableAlterColumn],
@@ -523,12 +535,16 @@ class SchemaDiffer(PydanticModel):
         current_type = exp.DataType.build(current_type, copy=False)
         if self.support_nested_operations:
             if new_type.this == current_type.this == exp.DataType.Type.STRUCT:
-                return self._get_operations(
-                    columns,
-                    current_type,
-                    new_type,
-                    root_struct,
-                )
+                if self.support_nested_drop or not self._requires_drop_alteration(
+                    current_type, new_type
+                ):
+                    return self._get_operations(
+                        columns,
+                        current_type,
+                        new_type,
+                        root_struct,
+                    )
+
             if new_type.this == current_type.this == exp.DataType.Type.ARRAY:
                 # Some engines (i.e. Snowflake) don't support defining types on arrays
                 if not new_type.expressions or not current_type.expressions:
@@ -536,12 +552,15 @@ class SchemaDiffer(PydanticModel):
                 new_array_type = new_type.expressions[0]
                 current_array_type = current_type.expressions[0]
                 if new_array_type.this == current_array_type.this == exp.DataType.Type.STRUCT:
-                    return self._get_operations(
-                        columns,
-                        current_array_type,
-                        new_array_type,
-                        root_struct,
-                    )
+                    if self.support_nested_drop or not self._requires_drop_alteration(
+                        current_array_type, new_array_type
+                    ):
+                        return self._get_operations(
+                            columns,
+                            current_array_type,
+                            new_array_type,
+                            root_struct,
+                        )
         if self._is_coerceable_type(current_type, new_type):
             return []
         elif self._is_compatible_type(current_type, new_type):
diff --git a/sqlmesh/core/snapshot/cache.py b/sqlmesh/core/snapshot/cache.py
index af573a80..0ace6f14 100644
--- a/sqlmesh/core/snapshot/cache.py
+++ b/sqlmesh/core/snapshot/cache.py
@@ -3,7 +3,12 @@ from __future__ import annotations
 import typing as t
 
 from pathlib import Path
-from sqlmesh.core.model.cache import OptimizedQueryCache
+from sqlmesh.core.model.cache import (
+    OptimizedQueryCache,
+    optimized_query_cache_pool,
+    load_optimized_query_cache,
+)
+from sqlmesh.core import constants as c
 from sqlmesh.core.snapshot.definition import Snapshot, SnapshotId
 from sqlmesh.utils.cache import FileCache
 
@@ -31,12 +36,10 @@ class SnapshotCache:
         """
         snapshots = {}
         cache_hits: t.Set[SnapshotId] = set()
+
         for s_id in snapshot_ids:
             snapshot = self._snapshot_cache.get(self._entry_name(s_id))
             if snapshot:
-                if snapshot.is_model:
-                    self._optimized_query_cache.with_optimized_query(snapshot.model)
-                self._update_node_hash_cache(snapshot)
                 snapshot.intervals = []
                 snapshot.dev_intervals = []
                 snapshots[s_id] = snapshot
@@ -46,18 +49,43 @@ class SnapshotCache:
         if snapshot_ids_to_load:
             loaded_snapshots = loader(snapshot_ids_to_load)
             for snapshot in loaded_snapshots:
-                self._update_node_hash_cache(snapshot)
-                self.put(snapshot)
                 snapshots[snapshot.snapshot_id] = snapshot
 
+        if c.MAX_FORK_WORKERS != 1:
+            with optimized_query_cache_pool(self._optimized_query_cache) as executor:
+                for key, entry_name in executor.map(
+                    load_optimized_query_cache,
+                    (
+                        (snapshot.model, s_id)
+                        for s_id, snapshot in snapshots.items()
+                        if snapshot.is_model
+                    ),
+                ):
+                    if entry_name:
+                        self._optimized_query_cache.with_optimized_query(
+                            snapshots[key].model, entry_name
+                        )
+
+        for snapshot in snapshots.values():
+            self._update_node_hash_cache(snapshot)
+
+            if snapshot.is_model and c.MAX_FORK_WORKERS == 1:
+                self._optimized_query_cache.with_optimized_query(snapshot.model)
+
+            self.put(snapshot)
+
         return snapshots, cache_hits
 
     def put(self, snapshot: Snapshot) -> None:
+        entry_name = self._entry_name(snapshot.snapshot_id)
+
+        if self._snapshot_cache.exists(entry_name):
+            return
+
         if snapshot.is_model:
-            self._optimized_query_cache.put(snapshot.model)
             # make sure we preload full_depends_on
             snapshot.model.full_depends_on
-        self._snapshot_cache.put(self._entry_name(snapshot.snapshot_id), value=snapshot)
+        self._snapshot_cache.put(entry_name, value=snapshot)
 
     def clear(self) -> None:
         self._snapshot_cache.clear()
diff --git a/sqlmesh/magics.py b/sqlmesh/magics.py
index 51652fc0..d5fb095b 100644
--- a/sqlmesh/magics.py
+++ b/sqlmesh/magics.py
@@ -325,8 +325,9 @@ class SQLMeshMagics(Magics):
     )
     @argument(
         "--skip-backfill",
+        "--dry-run",
         action="store_true",
-        help="Skip the backfill step.",
+        help="Skip the backfill step and only create a virtual update for the plan.",
     )
     @argument(
         "--forward-only",
@@ -734,12 +735,18 @@ class SQLMeshMagics(Magics):
         type=int,
         help="The max number of characters in a segment before creating new lines in pretty mode.",
     )
+    @argument(
+        "--check",
+        action="store_true",
+        help="Whether or not to check formatting (but not actually format anything).",
+        default=None,
+    )
     @line_magic
     @pass_sqlmesh_context
-    def format(self, context: Context, line: str) -> None:
+    def format(self, context: Context, line: str) -> bool:
         """Format all SQL models and audits."""
         args = parse_argstring(self.format, line)
-        context.format(**{k: v for k, v in vars(args).items() if v is not None})
+        return context.format(**{k: v for k, v in vars(args).items() if v is not None})
 
     @magic_arguments()
     @argument("environment", type=str, help="The environment to diff local state against.")


Test patch:
diff --git a/pytest.ini b/pytest.ini
index 3fd6e6ba..f1731df7 100644
--- a/pytest.ini
+++ b/pytest.ini
@@ -7,6 +7,7 @@ markers =
     docker: test that involves interacting with a Docker container
     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
 
     # Test Domain Markers
     # default: core functionality
diff --git a/tests/__init__.py b/tests/__init__.py
index d64b8864..2b35b8c9 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -1,3 +1,5 @@
+from sqlmesh.core import constants as c
 from sqlmesh.core.analytics import disable_analytics
 
+c.MAX_FORK_WORKERS = 1
 disable_analytics()
diff --git a/tests/cli/test_cli.py b/tests/cli/test_cli.py
index 309727c0..823b557c 100644
--- a/tests/cli/test_cli.py
+++ b/tests/cli/test_cli.py
@@ -214,13 +214,12 @@ def test_plan_restate_model(runner, tmp_path):
     assert_backfill_success(result)
 
 
-def test_plan_skip_backfill(runner, tmp_path):
+@pytest.mark.parametrize("flag", ["--skip-backfill", "--dry-run"])
+def test_plan_skip_backfill(runner, tmp_path, flag):
     create_example_project(tmp_path)
 
     # plan for `prod` errors if `--skip-backfill` is passed without --no-gaps
-    result = runner.invoke(
-        cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "--skip-backfill"]
-    )
+    result = runner.invoke(cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", flag])
     assert result.exit_code == 1
     assert (
         "Error: When targeting the production environment either the backfill should not be skipped or the lack of data gaps should be enforced (--no-gaps flag)."
@@ -231,7 +230,7 @@ def test_plan_skip_backfill(runner, tmp_path):
     # Input: `y` to perform virtual update
     result = runner.invoke(
         cli,
-        ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "--skip-backfill", "--no-gaps"],
+        ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", flag, "--no-gaps"],
         input="y\n",
     )
     assert result.exit_code == 0
@@ -562,15 +561,16 @@ def test_run_no_prod(runner, tmp_path):
     assert "Error: Environment 'prod' was not found." in result.output
 
 
+@pytest.mark.parametrize("flag", ["--skip-backfill", "--dry-run"])
 @freeze_time(FREEZE_TIME)
-def test_run_dev(runner, tmp_path):
+def test_run_dev(runner, tmp_path, flag):
     create_example_project(tmp_path)
 
     # Create dev environment but DO NOT backfill
     # Input: `y` for virtual update
     runner.invoke(
         cli,
-        ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "dev", "--skip-backfill"],
+        ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "dev", flag],
         input="y\n",
     )
 
diff --git a/tests/conftest.py b/tests/conftest.py
index 174e2a66..214a34fc 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -179,7 +179,7 @@ class SushiDataValidator:
 
 
 def pytest_collection_modifyitems(items, *args, **kwargs):
-    test_type_markers = {"fast", "slow", "docker", "remote"}
+    test_type_markers = {"fast", "slow", "docker", "remote", "isolated"}
     for item in items:
         for marker in item.iter_markers():
             if marker.name in test_type_markers:
diff --git a/tests/core/engine_adapter/test_base.py b/tests/core/engine_adapter/test_base.py
index b95d36e2..863e5793 100644
--- a/tests/core/engine_adapter/test_base.py
+++ b/tests/core/engine_adapter/test_base.py
@@ -474,6 +474,7 @@ def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture)
             {
                 "support_positional_add": True,
                 "support_nested_operations": True,
+                "support_nested_drop": True,
                 "array_element_selector": "element",
             },
             {
@@ -630,6 +631,7 @@ def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture)
             {
                 "support_positional_add": True,
                 "support_nested_operations": True,
+                "support_nested_drop": True,
                 "array_element_selector": "element",
             },
             {
@@ -659,6 +661,7 @@ def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture)
             {
                 "support_positional_add": True,
                 "support_nested_operations": True,
+                "support_nested_drop": True,
                 "array_element_selector": "element",
             },
             {
@@ -715,6 +718,7 @@ def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture)
         (
             {
                 "support_nested_operations": True,
+                "support_nested_drop": True,
                 "array_element_selector": "element",
             },
             {
@@ -772,6 +776,7 @@ def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture)
         (
             {
                 "support_nested_operations": True,
+                "support_nested_drop": True,
                 "array_element_selector": "element",
             },
             {
diff --git a/tests/core/test_format.py b/tests/core/test_format.py
index 907e0b4f..a75d8a5e 100644
--- a/tests/core/test_format.py
+++ b/tests/core/test_format.py
@@ -53,9 +53,14 @@ def test_format_files(tmp_path: pathlib.Path):
         == 'SELECT * FROM @this_model WHERE "CaseSensitive_item_id" < 0'
     )
 
+    assert not context.format(check=True)
+
     # Transpile project to BigQuery
     context.format(transpile="bigquery")
 
+    # Ensure format check is successful
+    assert context.format(transpile="bigquery", check=True)
+
     # Ensure transpilation success AND model specific dialect is mutated
     upd1 = f1.read_text(encoding="utf-8")
     assert (
diff --git a/tests/core/test_model.py b/tests/core/test_model.py
index 2d013abd..f02fd46c 100644
--- a/tests/core/test_model.py
+++ b/tests/core/test_model.py
@@ -283,7 +283,7 @@ def test_model_qualification():
         )
 
         model = load_sql_based_model(expressions)
-        model.render_query(optimize=True)
+        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"
@@ -2575,7 +2575,7 @@ def test_update_schema():
 
     logger = logging.getLogger("sqlmesh.core.renderer")
     with patch.object(logger, "warning") as mock_logger:
-        model.render_query(optimize=True)
+        model.render_query(needs_optimization=True)
         assert mock_logger.call_args[0][0] == missing_schema_warning_msg(
             '"db"."table"', ('"table_b"',)
         )
@@ -2586,7 +2586,7 @@ def test_update_schema():
         '"table_a"': {"a": "INT"},
         '"table_b"': {"b": "INT"},
     }
-    model.render_query(optimize=True)
+    model.render_query(needs_optimization=True)
 
 
 def test_missing_schema_warnings():
@@ -2609,27 +2609,27 @@ def test_missing_schema_warnings():
     # star, no schema, no deps
     with patch.object(logger, "warning") as mock_logger:
         model = load_sql_based_model(d.parse("MODEL (name test); SELECT * FROM (SELECT 1 a) x"))
-        model.render_query(optimize=True)
+        model.render_query(needs_optimization=True)
         mock_logger.assert_not_called()
 
     # star, full schema
     with patch.object(logger, "warning") as mock_logger:
         model = load_sql_based_model(d.parse("MODEL (name test); SELECT * FROM a CROSS JOIN b"))
         model.update_schema(full_schema)
-        model.render_query(optimize=True)
+        model.render_query(needs_optimization=True)
         mock_logger.assert_not_called()
 
     # star, partial schema
     with patch.object(logger, "warning") as mock_logger:
         model = load_sql_based_model(d.parse("MODEL (name test); SELECT * FROM a CROSS JOIN b"))
         model.update_schema(partial_schema)
-        model.render_query(optimize=True)
+        model.render_query(needs_optimization=True)
         assert mock_logger.call_args[0][0] == missing_schema_warning_msg('"test"', ('"b"',))
 
     # star, no schema
     with patch.object(logger, "warning") as mock_logger:
         model = load_sql_based_model(d.parse("MODEL (name test); SELECT * FROM b JOIN a"))
-        model.render_query(optimize=True)
+        model.render_query(needs_optimization=True)
         assert mock_logger.call_args[0][0] == missing_schema_warning_msg('"test"', ('"a"', '"b"'))
 
     # no star, full schema
@@ -2638,7 +2638,7 @@ def test_missing_schema_warnings():
             d.parse("MODEL (name test); SELECT x::INT FROM a CROSS JOIN b")
         )
         model.update_schema(full_schema)
-        model.render_query(optimize=True)
+        model.render_query(needs_optimization=True)
         mock_logger.assert_not_called()
 
     # no star, partial schema
@@ -2647,7 +2647,7 @@ def test_missing_schema_warnings():
             d.parse("MODEL (name test); SELECT x::INT FROM a CROSS JOIN b")
         )
         model.update_schema(partial_schema)
-        model.render_query(optimize=True)
+        model.render_query(needs_optimization=True)
         mock_logger.assert_not_called()
 
     # no star, empty schema
@@ -2655,7 +2655,7 @@ def test_missing_schema_warnings():
         model = load_sql_based_model(
             d.parse("MODEL (name test); SELECT x::INT FROM a CROSS JOIN b")
         )
-        model.render_query(optimize=True)
+        model.render_query(needs_optimization=True)
         mock_logger.assert_not_called()
 
 
@@ -4564,6 +4564,32 @@ def test_variables_python_sql_model(mocker: MockerFixture) -> None:
     )
 
 
+def test_columns_python_sql_model() -> None:
+    @model(
+        "test_columns_python_model",
+        is_sql=True,
+        kind="full",
+        columns={"d": "Date", "s": "String", "dt": "DateTime"},
+    )
+    def model_with_columns(evaluator, **kwargs):
+        return exp.select("*").from_("fake")
+
+    python_sql_model = model.get_registry()["test_columns_python_model"].model(
+        module_path=Path("."),
+        path=Path("."),
+    )
+
+    columns_to_types = python_sql_model.columns_to_types
+
+    assert columns_to_types is not None
+    assert isinstance(columns_to_types["d"], exp.DataType)
+    assert columns_to_types["d"].this == exp.DataType.Type.DATE
+    assert isinstance(columns_to_types["s"], exp.DataType)
+    assert columns_to_types["s"].this == exp.DataType.Type.TEXT
+    assert isinstance(columns_to_types["dt"], exp.DataType)
+    assert columns_to_types["dt"].this == exp.DataType.Type.DATETIME
+
+
 def test_named_variables_python_model(mocker: MockerFixture) -> None:
     @model(
         "test_named_variables_python_model",
@@ -5747,41 +5773,3 @@ def test_cache():
     model = load_sql_based_model(expressions)
     assert model.depends_on == {'"y"'}
     assert model.copy(update={"depends_on_": {'"z"'}}).depends_on == {'"z"', '"y"'}
-
-
-def test_parallel_load(assert_exp_eq, mocker):
-    import os
-    from sqlmesh.core import loader
-
-    pytest_current_test = os.environ.pop("PYTEST_CURRENT_TEST")
-    try:
-        spy = mocker.spy(loader, "_update_model_schemas_parallel")
-        context = Context(paths="examples/sushi")
-
-        if hasattr(os, "fork"):
-            spy.assert_called()
-
-        assert_exp_eq(
-            context.render("sushi.customers"),
-            """
-    WITH "current_marketing" AS (
-      SELECT
-        "marketing"."customer_id" AS "customer_id",
-        "marketing"."status" AS "status"
-      FROM "memory"."sushi"."marketing" AS "marketing"
-      WHERE
-        "marketing"."valid_to" IS NULL
-    )
-    SELECT DISTINCT
-      CAST("o"."customer_id" AS INT) AS "customer_id", /* this comment should not be registered */
-      "m"."status" AS "status",
-      "d"."zip" AS "zip"
-    FROM "memory"."sushi"."orders" AS "o"
-    LEFT JOIN "current_marketing" AS "m"
-      ON "m"."customer_id" = "o"."customer_id"
-    LEFT JOIN "memory"."raw"."demographics" AS "d"
-      ON "d"."customer_id" = "o"."customer_id"
-            """,
-        )
-    finally:
-        os.environ["PYTEST_CURRENT_TEST"] = pytest_current_test
diff --git a/tests/core/test_schema_diff.py b/tests/core/test_schema_diff.py
index f5581c58..43c0cbc4 100644
--- a/tests/core/test_schema_diff.py
+++ b/tests/core/test_schema_diff.py
@@ -407,7 +407,11 @@ def test_schema_diff_calculate_type_transitions():
                     "INT",
                 ),
             ],
-            dict(support_positional_add=True, support_nested_operations=True),
+            dict(
+                support_positional_add=True,
+                support_nested_operations=True,
+                support_nested_drop=True,
+            ),
         ),
         # Remove a column from the end of a struct
         (
@@ -423,7 +427,11 @@ def test_schema_diff_calculate_type_transitions():
                     "INT",
                 ),
             ],
-            dict(support_positional_add=True, support_nested_operations=True),
+            dict(
+                support_positional_add=True,
+                support_nested_operations=True,
+                support_nested_drop=True,
+            ),
         ),
         # Remove a column from the middle of a struct
         (
@@ -439,7 +447,36 @@ def test_schema_diff_calculate_type_transitions():
                     "INT",
                 ),
             ],
-            dict(support_positional_add=True, support_nested_operations=True),
+            dict(
+                support_positional_add=True,
+                support_nested_operations=True,
+                support_nested_drop=True,
+            ),
+        ),
+        # Remove a column from a struct where nested drop is not supported
+        (
+            "STRUCT<id INT, info STRUCT<col_a INT, col_b INT, col_c INT>>",
+            "STRUCT<id INT, info STRUCT<col_a INT, col_b INT>>",
+            [
+                TableAlterOperation.drop(
+                    [
+                        TableAlterColumn.struct("info"),
+                    ],
+                    expected_table_struct="STRUCT<id INT>",
+                    column_type="STRUCT<col_a INT, col_b INT, col_c INT>",
+                ),
+                TableAlterOperation.add(
+                    [
+                        TableAlterColumn.struct("info"),
+                    ],
+                    expected_table_struct="STRUCT<id INT, info STRUCT<col_a INT, col_b INT>>",
+                    column_type="STRUCT<col_a INT, col_b INT>",
+                ),
+            ],
+            dict(
+                support_nested_operations=True,
+                support_nested_drop=False,
+            ),
         ),
         # Remove two columns from the start of a struct
         (
@@ -463,7 +500,11 @@ def test_schema_diff_calculate_type_transitions():
                     "INT",
                 ),
             ],
-            dict(support_positional_add=True, support_nested_operations=True),
+            dict(
+                support_positional_add=True,
+                support_nested_operations=True,
+                support_nested_drop=True,
+            ),
         ),
         # Change a column type in a struct
         (
@@ -534,6 +575,35 @@ def test_schema_diff_calculate_type_transitions():
             dict(
                 support_positional_add=True,
                 support_nested_operations=True,
+                support_nested_drop=True,
+                compatible_types={
+                    exp.DataType.build("INT"): {exp.DataType.build("TEXT")},
+                },
+            ),
+        ),
+        # Add, remove and change a column from a struct where nested drop is not supported
+        (
+            "STRUCT<id INT, info STRUCT<col_a INT, col_b INT, col_c INT>>",
+            "STRUCT<id INT, info STRUCT<col_a INT, col_b TEXT, col_d INT>>",
+            [
+                TableAlterOperation.drop(
+                    [
+                        TableAlterColumn.struct("info"),
+                    ],
+                    expected_table_struct="STRUCT<id INT>",
+                    column_type="STRUCT<col_a INT, col_b INT, col_c INT>",
+                ),
+                TableAlterOperation.add(
+                    [
+                        TableAlterColumn.struct("info"),
+                    ],
+                    expected_table_struct="STRUCT<id INT, info STRUCT<col_a INT, col_b TEXT, col_d INT>>",
+                    column_type="STRUCT<col_a INT, col_b TEXT, col_d INT>",
+                ),
+            ],
+            dict(
+                support_nested_operations=True,
+                support_nested_drop=False,
                 compatible_types={
                     exp.DataType.build("INT"): {exp.DataType.build("TEXT")},
                 },
@@ -581,7 +651,11 @@ def test_schema_diff_calculate_type_transitions():
                     position=TableAlterColumnPosition.last("nest_col_a"),
                 ),
             ],
-            dict(support_positional_add=True, support_nested_operations=True),
+            dict(
+                support_positional_add=True,
+                support_nested_operations=True,
+                support_nested_drop=True,
+            ),
         ),
         # #####################
         # # Array Struct Tests
@@ -617,7 +691,11 @@ def test_schema_diff_calculate_type_transitions():
                     "INT",
                 ),
             ],
-            dict(support_positional_add=True, support_nested_operations=True),
+            dict(
+                support_positional_add=True,
+                support_nested_operations=True,
+                support_nested_drop=True,
+            ),
         ),
         # Alter column type in array of structs
         (
@@ -754,6 +832,7 @@ def test_schema_diff_calculate_type_transitions():
             dict(
                 support_positional_add=True,
                 support_nested_operations=True,
+                support_nested_drop=True,
             ),
         ),
         # Type with precision to same type with no precision and no default is DROP/ADD
diff --git a/tests/test_forking.py b/tests/test_forking.py
new file mode 100644
index 00000000..2ce48d39
--- /dev/null
+++ b/tests/test_forking.py
@@ -0,0 +1,42 @@
+import os
+import pytest
+
+from sqlmesh import Context
+from sqlmesh.core import loader
+
+
+pytestmark = pytest.mark.isolated
+
+
+def test_parallel_load(assert_exp_eq, mocker):
+    mocker.patch("sqlmesh.core.constants.MAX_FORK_WORKERS", 2)
+    spy = mocker.spy(loader, "_update_model_schemas_parallel")
+    context = Context(paths="examples/sushi")
+
+    if hasattr(os, "fork"):
+        spy.assert_called()
+
+    assert_exp_eq(
+        context.render("sushi.customers"),
+        """
+WITH "current_marketing" AS (
+  SELECT
+    "marketing"."customer_id" AS "customer_id",
+    "marketing"."status" AS "status"
+  FROM "memory"."sushi"."marketing" AS "marketing"
+  WHERE
+    "marketing"."valid_to" IS NULL
+)
+SELECT DISTINCT
+  CAST("o"."customer_id" AS INT) AS "customer_id", /* this comment should not be registered */
+  "m"."status" AS "status",
+  "d"."zip" AS "zip"
+FROM "memory"."sushi"."orders" AS "o"
+LEFT JOIN "current_marketing" AS "m"
+  ON "m"."customer_id" = "o"."customer_id"
+LEFT JOIN "memory"."raw"."demographics" AS "d"
+  ON "d"."customer_id" = "o"."customer_id"
+        """,
+    )
+
+    context.plan(no_prompts=True, auto_apply=True)