|
7 | 7 | import typing as t |
8 | 8 | import shutil |
9 | 9 | from datetime import datetime, timedelta |
| 10 | +import logging |
10 | 11 |
|
11 | 12 | import numpy as np # noqa: TID253 |
12 | 13 | import pandas as pd # noqa: TID253 |
13 | 14 | import pytest |
14 | 15 | import pytz |
| 16 | +from unittest import mock |
15 | 17 | from sqlglot import exp, parse_one |
16 | 18 | from sqlglot.optimizer.normalize_identifiers import normalize_identifiers |
17 | 19 | from sqlglot.optimizer.qualify_columns import quote_identifiers |
@@ -3697,3 +3699,65 @@ def _set_config(gateway: str, config: Config) -> None: |
3697 | 3699 | "incremental_model", |
3698 | 3700 | "seed_model", |
3699 | 3701 | ] |
| 3702 | + |
| 3703 | + |
| 3704 | +def test_materialized_view_evaluation(ctx: TestContext, mocker: MockerFixture): |
| 3705 | + adapter = ctx.engine_adapter |
| 3706 | + dialect = ctx.dialect |
| 3707 | + |
| 3708 | + if not adapter.SUPPORTS_MATERIALIZED_VIEWS: |
| 3709 | + pytest.skip(f"Skipping engine {dialect} as it does not support materialized views") |
| 3710 | + elif dialect in ("snowflake", "databricks"): |
| 3711 | + pytest.skip(f"Skipping {dialect} as they're not enabled on standard accounts") |
| 3712 | + |
| 3713 | + model_name = ctx.table("test_tbl") |
| 3714 | + mview_name = ctx.table("test_mview") |
| 3715 | + |
| 3716 | + sqlmesh = ctx.create_context() |
| 3717 | + |
| 3718 | + sqlmesh.upsert_model( |
| 3719 | + load_sql_based_model( |
| 3720 | + d.parse( |
| 3721 | + f""" |
| 3722 | + MODEL (name {model_name}, kind FULL); |
| 3723 | +
|
| 3724 | + SELECT 1 AS col |
| 3725 | + """ |
| 3726 | + ) |
| 3727 | + ) |
| 3728 | + ) |
| 3729 | + |
| 3730 | + sqlmesh.upsert_model( |
| 3731 | + load_sql_based_model( |
| 3732 | + d.parse( |
| 3733 | + f""" |
| 3734 | + MODEL (name {mview_name}, kind VIEW (materialized true)); |
| 3735 | +
|
| 3736 | + SELECT * FROM {model_name} |
| 3737 | + """ |
| 3738 | + ) |
| 3739 | + ) |
| 3740 | + ) |
| 3741 | + |
| 3742 | + def _assert_mview_value(value: int): |
| 3743 | + df = adapter.fetchdf(f"SELECT * FROM {mview_name.sql(dialect=dialect)}") |
| 3744 | + assert df["col"][0] == value |
| 3745 | + |
| 3746 | + # Case 1: Ensure that plan is successful and we can query the materialized view |
| 3747 | + sqlmesh.plan(auto_apply=True, no_prompts=True) |
| 3748 | + |
| 3749 | + _assert_mview_value(value=1) |
| 3750 | + |
| 3751 | + # Case 2: Ensure that we can change the underlying table and the materialized view is recreated |
| 3752 | + sqlmesh.upsert_model( |
| 3753 | + load_sql_based_model(d.parse(f"""MODEL (name {model_name}, kind FULL); SELECT 2 AS col""")) |
| 3754 | + ) |
| 3755 | + |
| 3756 | + logger = logging.getLogger("sqlmesh.core.snapshot.evaluator") |
| 3757 | + |
| 3758 | + with mock.patch.object(logger, "info") as mock_logger: |
| 3759 | + sqlmesh.plan(auto_apply=True, no_prompts=True) |
| 3760 | + |
| 3761 | + assert any("Replacing view" in call[0][0] for call in mock_logger.call_args_list) |
| 3762 | + |
| 3763 | + _assert_mview_value(value=2) |
0 commit comments