Merge branch 'main' into fix/clickhouse-cloud-mode-option

Author: cookieark7

sqlmesh/core/node.py

@@ -28,7 +28,8 @@
 class IntervalUnit(str, Enum):
     """IntervalUnit is the inferred granularity of an incremental node.
 
-    IntervalUnit can be one of 5 types, YEAR, MONTH, DAY, HOUR, MINUTE. The unit is inferred
+    IntervalUnit can be one of 7 types: YEAR, MONTH, DAY, HOUR, HALF_HOUR, QUARTER_HOUR,
+    FIVE_MINUTE. The unit is inferred
     based on the cron schedule of a node. The minimum time delta between a sample set of dates
     is used to determine which unit a node's schedule is.
 

sqlmesh/core/state_sync/db/interval.py

@@ -108,6 +108,9 @@ def remove_intervals(
                 for snapshot in all_snapshots
             ]
 
+        if not intervals_to_remove:
+            return
+
         if logger.isEnabledFor(logging.INFO):
             snapshot_ids = ", ".join(str(s.snapshot_id) for s, _ in intervals_to_remove)
             logger.info("Removing interval for snapshots: %s", snapshot_ids)

sqlmesh/core/test/runner.py

@@ -125,25 +125,7 @@ def run_tests(
     # Ensure workers are not greater than the number of tests
     num_workers = min(len(model_test_metadata) or 1, default_test_connection.concurrent_tasks)
 
-    def _run_single_test(
-        metadata: ModelTestMetadata, engine_adapter: EngineAdapter
-    ) -> t.Optional[ModelTextTestResult]:
-        test = ModelTest.create_test(
-            body=metadata.body,
-            test_name=metadata.test_name,
-            models=models,
-            engine_adapter=engine_adapter,
-            dialect=dialect,
-            path=metadata.path,
-            default_catalog=default_catalog,
-            preserve_fixtures=preserve_fixtures,
-            concurrency=num_workers > 1,
-            verbosity=verbosity,
-        )
-
-        if not test:
-            return None
-
+    def _run_single_test(test: ModelTest) -> ModelTextTestResult:
         result = t.cast(
             ModelTextTestResult,
             ModelTextTestRunner().run(t.cast(unittest.TestCase, test)),
@@ -158,11 +140,30 @@ def _run_single_test(
 
     start_time = time.perf_counter()
     try:
+        # Build ModelTest instances on the calling thread before workers start. create_test()
+        # can call to_datetime() / ttl_cache (time.time()), which races with another worker's
+        # time_machine freeze when execution_time is set under concurrent_tasks > 1.
+        # NOTE: We can run create_tests in a separate parallel stage for a future optimization.
+        # We just can't overlap runs/creations.
+        tests: list[ModelTest] = []
+        for metadata, engine_adapter in metadata_to_adapter.items():
+            test = ModelTest.create_test(
+                body=metadata.body,
+                test_name=metadata.test_name,
+                models=models,
+                engine_adapter=engine_adapter,
+                dialect=dialect,
+                path=metadata.path,
+                default_catalog=default_catalog,
+                preserve_fixtures=preserve_fixtures,
+                concurrency=num_workers > 1,
+                verbosity=verbosity,
+            )
+            if test:
+                tests.append(test)
+
         with ThreadPoolExecutor(max_workers=num_workers) as pool:
-            futures = [
-                pool.submit(_run_single_test, metadata=metadata, engine_adapter=engine_adapter)
-                for metadata, engine_adapter in metadata_to_adapter.items()
-            ]
+            futures = [pool.submit(_run_single_test, test) for test in tests]
 
             for future in concurrent.futures.as_completed(futures):
                 test_results.append(future.result())

tests/core/integration/test_restatement.py

@@ -83,6 +83,44 @@ def test_restatement_plan_ignores_changes(init_and_plan_context: t.Callable):
     context.apply(plan)
 
 
+@time_machine.travel("2023-01-08 15:00:00 UTC")
+def test_prod_restatement_with_unbackfilled_dev_version(init_and_plan_context: t.Callable):
+    """
+    Scenario:
+        Prod is built. A breaking change is planned to `dev` with `--skip-backfill`,
+        so `dev` holds a different snapshot version with no interval rows. Prod is
+        then restated.
+    Outcome:
+        RestatementStage tries to clear `dev` intervals for that other version, finds
+        none in `_intervals`, and no-ops instead of crashing on an empty insert. Prod
+        restatement still applies; the un-backfilled `dev` snapshot still has no
+        intervals.
+    """
+    context, plan = init_and_plan_context("examples/sushi")
+    context.apply(plan)
+
+    prod_snapshot_id = context.get_snapshot("sushi.waiter_revenue_by_day").snapshot_id
+
+    context.upsert_model(
+        add_projection_to_model(t.cast(SqlModel, context.get_model("sushi.waiter_revenue_by_day")))
+    )
+    context.plan("dev", skip_backfill=True, auto_apply=True, no_prompts=True)
+
+    dev_snapshot_id = context.get_snapshot("sushi.waiter_revenue_by_day").snapshot_id
+    assert dev_snapshot_id != prod_snapshot_id
+    assert not context.state_sync.get_snapshots([dev_snapshot_id])[dev_snapshot_id].intervals
+
+    context.plan(
+        restate_models=["sushi.waiter_revenue_by_day"],
+        start="2023-01-07",
+        end="2023-01-08",
+        auto_apply=True,
+        no_prompts=True,
+    )
+
+    assert not context.state_sync.get_snapshots([dev_snapshot_id])[dev_snapshot_id].intervals
+
+
 @time_machine.travel("2023-01-08 15:00:00 UTC")
 def test_restatement_plan_across_environments_snapshot_with_shared_version(
     init_and_plan_context: t.Callable,

tests/core/state_sync/test_state_sync.py

@@ -400,6 +400,40 @@ def test_remove_interval_missing_snapshot(
     ]
 
 
+def test_remove_interval_no_matching_intervals(
+    state_sync: EngineAdapterStateSync, make_snapshot: t.Callable
+) -> None:
+    snapshot = make_snapshot(
+        SqlModel(
+            name="a",
+            cron="@daily",
+            query=parse_one("select 1, ds"),
+        ),
+        version="a",
+    )
+    state_sync.push_snapshots([snapshot])
+
+    # The snapshot has never been backfilled, so there are no rows to expand the shared versions from
+    state_sync.remove_intervals(
+        [(snapshot, snapshot.inclusive_exclusive("2020-01-15", "2020-01-17"))],
+        remove_shared_versions=True,
+    )
+
+    remove_records_count = state_sync.engine_adapter.fetchone(
+        "SELECT COUNT(*) FROM sqlmesh._intervals WHERE name = '\"a\"' AND version = 'a' AND is_removed"
+    )[0]  # type: ignore
+    assert remove_records_count == 0
+
+    assert not state_sync.get_snapshots([snapshot])[snapshot.snapshot_id].intervals
+
+
+def test_remove_interval_empty_input(state_sync: EngineAdapterStateSync) -> None:
+    state_sync.remove_intervals([])
+    state_sync.remove_intervals([], remove_shared_versions=True)
+
+    assert state_sync.engine_adapter.fetchone("SELECT COUNT(*) FROM sqlmesh._intervals")[0] == 0  # type: ignore
+
+
 def test_refresh_snapshot_intervals(
     state_sync: EngineAdapterStateSync, make_snapshot: t.Callable
 ) -> None: