Merge branch 'main' into fix/snowflake-iceberg-column-comments

Author: sineline

.devcontainer/devcontainer.json

@@ -11,10 +11,7 @@
   "postCreateCommand": "bash .devcontainer/post-create-command.sh",
   "customizations": {
     "vscode": {
-      "extensions": [
-        "ms-python.python",
-        "ms-python.vscode-pylance"
-      ]
+      "extensions": ["ms-python.python", "ms-python.vscode-pylance"]
     }
   },
   "remoteUser": "vscode"

.github/workflows/pr.yaml

@@ -16,6 +16,7 @@ jobs:
     outputs:
       python: ${{ steps.filter.outputs.python }}
       client: ${{ steps.filter.outputs.client }}
+      vscode: ${{ steps.filter.outputs.vscode }}
       ci: ${{ steps.filter.outputs.ci }}
     steps:
       - uses: actions/checkout@v7
@@ -34,6 +35,8 @@ jobs:
               - 'pyproject.toml'
             client:
               - 'web/client/**'
+            vscode:
+              - 'vscode/**'
             ci:
               - '.github/**'
               - 'Makefile'
@@ -188,9 +191,10 @@ jobs:
 
   ui-style:
     needs: [changes]
-    if: false
-      # needs.changes.outputs.client == 'true' || needs.changes.outputs.ci ==
-      # 'true' || github.ref == 'refs/heads/main'
+    if:
+      needs.changes.outputs.client == 'true' || needs.changes.outputs.vscode ==
+      'true' || needs.changes.outputs.ci == 'true' || github.ref ==
+      'refs/heads/main'
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v7
@@ -252,7 +256,17 @@ jobs:
       fail-fast: false
       matrix:
         engine:
-          [duckdb, postgres, mysql, mssql, trino, spark, clickhouse, risingwave, starrocks]
+          [
+            duckdb,
+            postgres,
+            mysql,
+            mssql,
+            trino,
+            spark,
+            clickhouse,
+            risingwave,
+            starrocks,
+          ]
     env:
       PYTEST_XDIST_AUTO_NUM_WORKERS: 2
       SQLMESH__DISABLE_ANONYMIZED_ANALYTICS: '1'
@@ -393,10 +407,13 @@ jobs:
           retention-days: 7
 
   test-vscode:
+    needs: changes
+    if:
+      needs.changes.outputs.vscode == 'true' || needs.changes.outputs.ci ==
+      'true' || github.ref == 'refs/heads/main'
     env:
       PLAYWRIGHT_SKIP_BROWSER_DOWNLOAD: 1
     runs-on: ubuntu-latest
-    if: false
     steps:
       - uses: actions/checkout@v7
       - uses: actions/setup-node@v7
@@ -457,7 +474,19 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        dbt-version: ['1.3', '1.4', '1.5', '1.6', '1.7', '1.8', '1.9', '1.10', '1.11', '1.12']
+        dbt-version:
+          [
+            '1.3',
+            '1.4',
+            '1.5',
+            '1.6',
+            '1.7',
+            '1.8',
+            '1.9',
+            '1.10',
+            '1.11',
+            '1.12',
+          ]
     steps:
       - uses: actions/checkout@v7
       - name: Set up Python

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/definition.py

@@ -263,12 +263,9 @@ def assert_equal(
         for col, value in object_sentinel_values.items():
             try:
                 # can't use `isinstance()` here - https://stackoverflow.com/a/68743663/1707525
-                if type(value) is datetime.date:
-                    expected[col] = pd.to_datetime(expected[col]).dt.date
-                elif type(value) is datetime.time:
-                    expected[col] = pd.to_datetime(expected[col]).dt.time
-                elif type(value) is datetime.datetime:
-                    expected[col] = pd.to_datetime(expected[col]).dt.to_pydatetime()
+                value_type = type(value)
+                if value_type in (datetime.date, datetime.time, datetime.datetime):
+                    expected[col] = _parse_expected_datetime_column(expected[col], value_type)
             except Exception as e:
                 from sqlmesh.core.console import get_console
 
@@ -1014,6 +1011,34 @@ def _raise_error(msg: str, path: Path | None = None) -> None:
     raise TestError(f"Failed to run test:\n{msg}")
 
 
+def _parse_expected_datetime_column(series: pd.Series, target_type: type) -> pd.Series:
+    """Convert a series of expected values to python ``date``/``time``/``datetime``.
+
+    Falls back to microsecond resolution when pandas' default nanosecond
+    parsing overflows. SQL ``TIMESTAMP`` columns can carry values outside
+    pandas' default ``datetime64[ns]`` range (1677-09-21..2262-04-11), so
+    unit tests may compare against values like ``0001-01-01`` which are
+    valid in the database but overflow the default resolution.
+    """
+    import pandas as pd
+    from pandas.errors import OutOfBoundsDatetime
+
+    try:
+        parsed = pd.to_datetime(series)
+    except OutOfBoundsDatetime:
+        parsed = series.astype("datetime64[us]")
+
+    if target_type is datetime.date:
+        return parsed.dt.date
+    if target_type is datetime.time:
+        return parsed.dt.time
+    # `Series.dt.to_pydatetime()` returns an `ndarray` in pandas 2.x. Wrap it in a
+    # Series with ``dtype=object`` so pandas does not coerce the values back to
+    # ``pd.Timestamp`` (which would reintroduce the nanosecond overflow this
+    # function exists to avoid).
+    return pd.Series(parsed.dt.to_pydatetime(), index=parsed.index, dtype="object")
+
+
 def _normalize_df_value(value: t.Any) -> t.Any:
     """Normalize data in a pandas dataframe so ruamel and sqlglot can deal with it."""
     import numpy as np

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:

tests/core/test_test.py

@@ -2931,6 +2931,48 @@ def test_timestamp_normalization() -> None:
     )
 
 
+def test_out_of_bounds_nanosecond_timestamp_comparison(mocker: MockerFixture) -> None:
+    # https://github.com/TobikoData/sqlmesh/issues/5929
+    # Engines like Redshift may return a TIMESTAMP column as an object-dtype
+    # series of python `datetime.datetime` instances. Values outside pandas'
+    # default `datetime64[ns]` range (1677-09-21..2262-04-11) - which SQL
+    # `TIMESTAMP` fully supports - previously raised `OutOfBoundsDatetime`
+    # while parsing the expected values, producing a "Failed to convert
+    # expected value into `datetime`" warning and a false mismatch on values
+    # whose repr survives str-coercion (the values below happen to compare
+    # equal via `str()`, so the mismatch was silent).
+    test = _create_test(
+        body=load_yaml(
+            """
+test_foo:
+  model: sushi.foo
+  outputs:
+    query:
+      - ts_col: "0001-01-01 00:00:00"
+      - ts_col: "9999-12-31 23:59:59"
+            """
+        ),
+        test_name="test_foo",
+        model=_create_model("SELECT ts_col FROM raw"),
+        context=Context(config=Config(model_defaults=ModelDefaultsConfig(dialect="duckdb"))),
+    )
+    actual = pd.DataFrame(
+        {
+            "ts_col": pd.Series(
+                [datetime.datetime(1, 1, 1), datetime.datetime(9999, 12, 31, 23, 59, 59)],
+                dtype=object,
+            )
+        }
+    )
+    # Use T separator so a string-only comparison (broken path) would mismatch
+    # against str(datetime.datetime(1, 1, 1)) == "0001-01-01 00:00:00".
+    expected = pd.DataFrame({"ts_col": ["0001-01-01T00:00:00", "9999-12-31T23:59:59"]})
+    log_warning = mocker.spy(get_console(), "log_warning")
+    test.assert_equal(expected=expected, actual=actual, sort=False)
+    for call_args in log_warning.call_args_list:
+        assert "Failed to convert expected value" not in call_args.args[0]
+
+
 @use_terminal_console
 def test_disable_test_logging_if_no_tests_found(mocker: MockerFixture, tmp_path: Path) -> None:
     init_example_project(tmp_path, engine_type="duckdb")

tests/web/test_models.py

@@ -0,0 +1,25 @@
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+import pytest
+
+from sqlmesh.core.context import Context
+from web.server.api.endpoints.models import get_models
+
+pytestmark = pytest.mark.web
+
+
+def test_get_models_multi_repo() -> None:
+    """Models of every project are serialized, not just those of the first one.
+
+    `context.path` is the first configured project, so it is not an ancestor of the models
+    defined in any of the others.
+    """
+    context = Context(paths=["examples/multi/repo_1", "examples/multi/repo_2"], gateway="memory")
+
+    paths_by_name = {model.name: model.path for model in get_models(context)}
+
+    # Each model is reported relative to the project that defines it.
+    assert paths_by_name["bronze.a"] == "models/a.sql"
+    assert paths_by_name["silver.c"] == "models/c.sql"