Seeds are now handled in evaluator

Author: treysp

sqlmesh/core/engine_adapter/databricks.py

@@ -403,8 +403,8 @@ def _record_execution_stats(
                 if history_df is not None and not history_df.empty:
                     write_df = history_df[history_df["operation"] == "WRITE"]
                     write_df = write_df[write_df["timestamp"] == write_df["timestamp"].max()]
-                    if not write_df.empty:
-                        metrics = write_df["operationMetrics"][0]
+                    if not write_df.empty and "operationMetrics" in write_df.columns:
+                        metrics = write_df["operationMetrics"].iloc[0]
                         if metrics:
                             rowcount = None
                             rowcount_str = [

sqlmesh/core/scheduler.py

@@ -540,21 +540,21 @@ def run_node(node: SchedulingUnit) -> None:
                     )
 
                     self.console.update_snapshot_evaluation_progress(
-                            snapshot,
-                            batched_intervals[snapshot][node.batch_index],
-                            node.batch_index,
-                            evaluation_duration_ms,
-                            num_audits - num_audits_failed,
-                            num_audits_failed,
-                            execution_stats=execution_stats,
+                        snapshot,
+                        batched_intervals[snapshot][node.batch_index],
+                        node.batch_index,
+                        evaluation_duration_ms,
+                        num_audits - num_audits_failed,
+                        num_audits_failed,
+                        execution_stats=execution_stats,
                     )
             elif isinstance(node, CreateNode):
                 self.snapshot_evaluator.create_snapshot(
                     snapshot=snapshot,
                     snapshots=self.snapshots_by_name,
                     deployability_index=deployability_index,
                     allow_destructive_snapshots=allow_destructive_snapshots or set(),
-                    )
+                )
 
         try:
             with self.snapshot_evaluator.concurrent_context():

tests/core/engine_adapter/integration/test_integration.py

@@ -2449,13 +2449,12 @@ def capture_execution_stats(
     assert len(physical_layer_results.tables) == len(physical_layer_results.non_temp_tables) == 3
 
     if ctx.engine_adapter.SUPPORTS_QUERY_EXECUTION_TRACKING:
+        assert actual_execution_stats["seed_model"].total_rows_processed == 7
         assert actual_execution_stats["incremental_model"].total_rows_processed == 7
         # snowflake doesn't track rows for CTAS
         assert actual_execution_stats["full_model"].total_rows_processed == (
             None if ctx.mark.startswith("snowflake") else 3
         )
-        # seed rows aren't tracked
-        assert actual_execution_stats["seed_model"].total_rows_processed is None
 
         if ctx.mark.startswith("bigquery") or ctx.mark.startswith("databricks"):
             assert actual_execution_stats["incremental_model"].total_bytes_processed is not None

tests/core/test_snapshot_evaluator.py

@@ -617,7 +617,6 @@ def test_evaluate_materialized_view_with_partitioned_by_cluster_by(
 
     execute_mock.assert_has_calls(
         [
-            call("CREATE SCHEMA IF NOT EXISTS `sqlmesh__test_schema`", False),
             call(
                 f"CREATE MATERIALIZED VIEW `sqlmesh__test_schema`.`test_schema__test_model__{snapshot.version}` PARTITION BY `a` CLUSTER BY `b` AS SELECT `a` AS `a`, `b` AS `b` FROM `tbl` AS `tbl`",
                 False,