Feat: Lazily create model tables during evaluation

Author: izeigerman

examples/sushi/config.py

@@ -27,6 +27,11 @@
 defaults = {"dialect": "duckdb"}
 model_defaults = ModelDefaultsConfig(**defaults)
 model_defaults_iceberg = ModelDefaultsConfig(**defaults, storage_format="iceberg")
+before_all = [
+    "CREATE SCHEMA IF NOT EXISTS raw",
+    "DROP VIEW IF EXISTS raw.demographics",
+    "CREATE VIEW raw.demographics AS (SELECT 1 AS customer_id, '00000' AS zip)",
+]
 
 
 # A DuckDB config, in-memory by default.
@@ -52,6 +57,7 @@
             "nomissingexternalmodels",
         ],
     ),
+    before_all=before_all,
 )
 
 bigquery_config = Config(
@@ -63,6 +69,7 @@
     },
     default_gateway="bq",
     model_defaults=model_defaults,
+    before_all=before_all,
 )
 
 # A configuration used for SQLMesh tests.
@@ -75,6 +82,7 @@
         )
     ),
     model_defaults=model_defaults,
+    before_all=before_all,
 )
 
 # A configuration used for SQLMesh tests with virtual environment mode set to DEV_ONLY.
@@ -84,14 +92,15 @@
         "plan": PlanConfig(
             auto_categorize_changes=CategorizerConfig.all_full(),
         ),
-    }
+    },
 )
 
 # A DuckDB config with a physical schema map.
 map_config = Config(
     default_connection=DuckDBConnectionConfig(),
     physical_schema_mapping={"^sushi$": "company_internal"},
     model_defaults=model_defaults,
+    before_all=before_all,
 )
 
 # A config representing isolated systems with a gateway per system
@@ -103,6 +112,7 @@
     },
     default_gateway="dev",
     model_defaults=model_defaults,
+    before_all=before_all,
 )
 
 required_approvers_config = Config(
@@ -137,19 +147,21 @@
         ),
     ],
     model_defaults=model_defaults,
+    before_all=before_all,
 )
 
 
 environment_suffix_table_config = Config(
     default_connection=DuckDBConnectionConfig(),
     model_defaults=model_defaults,
     environment_suffix_target=EnvironmentSuffixTarget.TABLE,
+    before_all=before_all,
 )
 
 environment_suffix_catalog_config = environment_suffix_table_config.model_copy(
     update={
         "environment_suffix_target": EnvironmentSuffixTarget.CATALOG,
-    }
+    },
 )
 
 CATALOGS = {
@@ -161,6 +173,7 @@
     default_connection=DuckDBConnectionConfig(catalogs=CATALOGS),
     default_test_connection=DuckDBConnectionConfig(catalogs=CATALOGS),
     model_defaults=model_defaults,
+    before_all=before_all,
 )
 
 environment_catalog_mapping_config = Config(
@@ -177,4 +190,5 @@
         "^prod$": "prod_catalog",
         ".*": "dev_catalog",
     },
+    before_all=before_all,
 )

sqlmesh/core/engine_adapter/base.py

@@ -443,14 +443,20 @@ def replace_query(
             target_table=target_table,
             source_columns=source_columns,
         )
+        if not target_columns_to_types and table_exists:
+            target_columns_to_types = self.columns(target_table)
         query = source_queries[0].query_factory()
-        target_columns_to_types = target_columns_to_types or self.columns(target_table)
         self_referencing = any(
             quote_identifiers(table) == quote_identifiers(target_table)
             for table in query.find_all(exp.Table)
         )
         # If a query references itself then it must have a table created regardless of approach used.
         if self_referencing:
+            if not target_columns_to_types:
+                raise SQLMeshError(
+                    f"Cannot create a self-referencing table {target_table.sql(dialect=self.dialect)} without knowing the column types. "
+                    "Try casting the columns to an expected type or defining the columns in the model metadata. "
+                )
             self._create_table_from_columns(
                 target_table,
                 target_columns_to_types,
@@ -472,6 +478,7 @@ def replace_query(
                 **kwargs,
             )
         if self_referencing:
+            assert target_columns_to_types is not None
             with self.temp_table(
                 self._select_columns(target_columns_to_types).from_(target_table),
                 name=target_table,

sqlmesh/core/engine_adapter/base_postgres.py

@@ -53,7 +53,9 @@ def columns(
         self.execute(sql)
         resp = self.cursor.fetchall()
         if not resp:
-            raise SQLMeshError("Could not get columns for table '%s'. Table not found.", table_name)
+            raise SQLMeshError(
+                f"Could not get columns for table '{table.sql(dialect=self.dialect)}'. Table not found."
+            )
         return {
             column_name: exp.DataType.build(data_type, dialect=self.dialect, udt=True)
             for column_name, data_type in resp

sqlmesh/core/plan/evaluator.py

@@ -205,6 +205,16 @@ def visit_physical_layer_update_stage(
                     success=completion_status is not None and completion_status.is_success
                 )
 
+    def visit_physical_layer_schema_creation_stage(
+        self, stage: stages.PhysicalLayerSchemaCreationStage, plan: EvaluatablePlan
+    ) -> None:
+        try:
+            self.snapshot_evaluator.create_physical_schemas(
+                stage.snapshots, stage.deployability_index
+            )
+        except Exception as ex:
+            raise PlanError("Plan application failed.") from ex
+
     def visit_backfill_stage(self, stage: stages.BackfillStage, plan: EvaluatablePlan) -> None:
         if plan.empty_backfill:
             intervals_to_add = []

sqlmesh/core/plan/stages.py

@@ -71,6 +71,19 @@ class PhysicalLayerUpdateStage:
     deployability_index: DeployabilityIndex
 
 
+@dataclass
+class PhysicalLayerSchemaCreationStage:
+    """Create the physical schemas for the given snapshots.
+
+    Args:
+        snapshots: Snapshots to create physical schemas for.
+        deployability_index: Deployability index for this stage.
+    """
+
+    snapshots: t.List[Snapshot]
+    deployability_index: DeployabilityIndex
+
+
 @dataclass
 class AuditOnlyRunStage:
     """Run audits only for given snapshots.
@@ -185,6 +198,7 @@ class FinalizeEnvironmentStage:
     AfterAllStage,
     CreateSnapshotRecordsStage,
     PhysicalLayerUpdateStage,
+    PhysicalLayerSchemaCreationStage,
     AuditOnlyRunStage,
     RestatementStage,
     BackfillStage,
@@ -283,11 +297,19 @@ def build(self, plan: EvaluatablePlan) -> t.List[PlanStage]:
         if plan.new_snapshots:
             stages.append(CreateSnapshotRecordsStage(snapshots=plan.new_snapshots))
 
-        stages.append(
-            self._get_physical_layer_update_stage(
-                plan, snapshots, snapshots_to_intervals, deployability_index_for_creation
+        if plan.skip_backfill or plan.empty_backfill:
+            stages.append(
+                self._get_physical_layer_update_stage(
+                    plan, snapshots, snapshots_to_intervals, deployability_index_for_creation
+                )
+            )
+        else:
+            stages.append(
+                PhysicalLayerSchemaCreationStage(
+                    snapshots=self._get_snapshots_to_create(plan, snapshots),
+                    deployability_index=deployability_index,
+                )
             )
-        )
 
         audit_only_snapshots = self._get_audit_only_snapshots(new_snapshots)
         if audit_only_snapshots: