Skip to content

Commit 55d2c86

Browse files
committed
Fix: render execution_time through a per-test generator, not a shared patch
ModelTest froze time by patching the dialect's generator TRANSFORMS and SQLGlot's cached dispatch table, which are process-global, under a lock that only covers the model and CTE renders. setUp (fixture CREATE VIEW) and tearDown render outside that lock, so with concurrent_tasks > 1 another test could observe the dispatch table mid-restore and fail with "Unsupported expression type Create", or silently render CURRENT_* at the other test's frozen time. Build the frozen-time transforms into a per-test generator subclass and render SQL model tests through it; keep the lock for time_machine only. Python model tests still patch the shared dialect under the lock because their SQL goes through the engine adapter. Signed-off-by: Niklas Dohmen <niklas@enam.co>
1 parent b1e36b9 commit 55d2c86

2 files changed

Lines changed: 146 additions & 31 deletions

File tree

sqlmesh/core/test/definition.py

Lines changed: 83 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@
3434
import pandas as pd
3535

3636
from sqlglot.dialects.dialect import DialectType
37+
from sqlglot.generator import Generator
3738

3839
Row = t.Dict[str, t.Any]
3940

@@ -48,6 +49,54 @@
4849
}
4950

5051

52+
_FROZEN_TIME_GENERATORS: t.Dict[
53+
t.Tuple[t.Type[Generator], str, t.Optional[str]], t.Type[Generator]
54+
] = {}
55+
_FROZEN_TIME_GENERATORS_LOCK = threading.Lock()
56+
57+
58+
def _frozen_time_generator_class(
59+
generator_class: t.Type[Generator], execution_time: str, dialect: t.Optional[str]
60+
) -> t.Type[Generator]:
61+
"""Returns a subclass of `generator_class` whose CURRENT_* transforms render `execution_time`.
62+
63+
SQLGlot caches one dispatch table per generator class, so a subclass gets its own table and
64+
the dialect's shared generator class is never modified. Subclasses are cached per
65+
(generator, execution time, dialect) so tests that share an execution time share a table.
66+
"""
67+
key = (generator_class, execution_time, dialect)
68+
with _FROZEN_TIME_GENERATORS_LOCK:
69+
klass = _FROZEN_TIME_GENERATORS.get(key)
70+
if klass is None:
71+
exec_time = exp.Literal.string(execution_time)
72+
klass = t.cast(
73+
t.Type["Generator"],
74+
type(
75+
f"{generator_class.__name__}FrozenTime",
76+
(generator_class,),
77+
{
78+
"TRANSFORMS": {
79+
**generator_class.TRANSFORMS,
80+
exp.CurrentDate: lambda self, _: self.sql(
81+
exp.cast(exec_time, "date", dialect=dialect)
82+
),
83+
exp.CurrentDatetime: lambda self, _: self.sql(
84+
exp.cast(exec_time, "datetime", dialect=dialect)
85+
),
86+
exp.CurrentTime: lambda self, _: self.sql(
87+
exp.cast(exec_time, "time", dialect=dialect)
88+
),
89+
exp.CurrentTimestamp: lambda self, _: self.sql(
90+
exp.cast(exec_time, "timestamp", dialect=dialect)
91+
),
92+
}
93+
},
94+
),
95+
)
96+
_FROZEN_TIME_GENERATORS[key] = klass
97+
return klass
98+
99+
51100
class ModelTest(unittest.TestCase):
52101
__test__ = False
53102

@@ -116,31 +165,22 @@ def __init__(
116165
)
117166
self._qualified_fixture_schema = schema_(self._fixture_schema, self._fixture_catalog)
118167

119-
self._transforms = self._test_adapter_dialect.generator_class.TRANSFORMS
120168
self._execution_time = str(self.body.get("vars", {}).get("execution_time") or "")
121169

122170
if self._execution_time:
123171
# Normalizes the execution time by converting it into UTC timezone
124172
self._execution_time = str(to_datetime(self._execution_time))
125173

126-
# When execution_time is set, we mock the CURRENT_* SQL expressions so they always return it
174+
# When execution_time is set, the CURRENT_* SQL expressions must render as that time. The
175+
# overrides live on a per-test generator subclass rather than on the dialect's shared
176+
# generator class: SQLGlot caches one dispatch table per generator class, so patching the
177+
# shared one is visible to every other thread that renders SQL (e.g. a concurrent test
178+
# creating its fixture views) and races with the patch's restoration.
179+
self._generator_class = self._test_adapter_dialect.generator_class
127180
if self._execution_time:
128-
exec_time = exp.Literal.string(self._execution_time)
129-
self._transforms = {
130-
**self._transforms,
131-
exp.CurrentDate: lambda self, _: self.sql(
132-
exp.cast(exec_time, "date", dialect=dialect)
133-
),
134-
exp.CurrentDatetime: lambda self, _: self.sql(
135-
exp.cast(exec_time, "datetime", dialect=dialect)
136-
),
137-
exp.CurrentTime: lambda self, _: self.sql(
138-
exp.cast(exec_time, "time", dialect=dialect)
139-
),
140-
exp.CurrentTimestamp: lambda self, _: self.sql(
141-
exp.cast(exec_time, "timestamp", dialect=dialect)
142-
),
143-
}
181+
self._generator_class = _frozen_time_generator_class(
182+
self._generator_class, self._execution_time, dialect
183+
)
144184

145185
super().__init__()
146186

@@ -603,15 +643,18 @@ def _normalize_column_name(self, name: str) -> str:
603643
return normalized_name
604644

605645
@contextmanager
606-
def _concurrent_render_context(self) -> t.Iterator[None]:
646+
def _concurrent_render_context(self, patch_shared_dialect: bool = False) -> t.Iterator[None]:
607647
"""
608648
Context manager that ensures that the tests are executed safely in a concurrent environment.
609-
This is needed in case `execution_time` is set, as we'd then have to:
610-
- Freeze time through `time_machine` (not thread safe)
611-
- Globally patch the SQLGlot dialect so that any date/time nodes are evaluated at the `execution_time` during generation
649+
This is needed in case `execution_time` is set, as we'd then have to freeze time through
650+
`time_machine`, which is not thread safe.
651+
652+
SQL model tests render through `self._generator_class`, so the shared dialect is never
653+
modified. Python model tests may run arbitrary SQL through the engine adapter, whose
654+
generator cannot be swapped per test, so they additionally patch the shared generator's
655+
transforms (`patch_shared_dialect=True`) while holding the lock.
612656
"""
613657
import time_machine
614-
from sqlglot.generator import _DISPATCH_CACHE
615658

616659
lock_ctx: AbstractContextManager = (
617660
self.CONCURRENT_RENDER_LOCK if self.concurrency else nullcontext()
@@ -621,19 +664,30 @@ def _concurrent_render_context(self) -> t.Iterator[None]:
621664
dispatch_patch_ctx: AbstractContextManager = nullcontext()
622665

623666
if self._execution_time:
624-
generator_class = self._test_adapter_dialect.generator_class
625667
time_ctx = time_machine.travel(self._execution_time, tick=False)
626-
dialect_patch_ctx = patch.dict(generator_class.TRANSFORMS, self._transforms)
668+
669+
if self._execution_time and patch_shared_dialect:
670+
from sqlglot.generator import _DISPATCH_CACHE
671+
672+
generator_class = self._test_adapter_dialect.generator_class
673+
transforms = self._generator_class.TRANSFORMS
674+
dialect_patch_ctx = patch.dict(generator_class.TRANSFORMS, transforms)
627675

628676
# sqlglot caches a dispatch table per generator class, so we need to patch
629677
# it as well to ensure the overridden transforms are actually used
630678
dispatch = _DISPATCH_CACHE.get(generator_class)
631679
if dispatch is not None:
632-
dispatch_patch_ctx = patch.dict(dispatch, self._transforms)
680+
dispatch_patch_ctx = patch.dict(dispatch, transforms)
633681

634682
with lock_ctx, time_ctx, dialect_patch_ctx, dispatch_patch_ctx:
635683
yield
636684

685+
def _generate_sql(self, expression: exp.Expr) -> str:
686+
"""Generates SQL for the testing engine, rendering CURRENT_* at `execution_time` when set."""
687+
return self._generator_class(
688+
dialect=self._test_adapter_dialect, pretty=self.engine_adapter._pretty_sql
689+
).generate(expression)
690+
637691
def _execute(self, query: exp.Query | str) -> pd.DataFrame:
638692
"""Executes the given query using the testing engine adapter and returns a DataFrame."""
639693
return self.engine_adapter.fetchdf(query)
@@ -701,9 +755,7 @@ def test_ctes(self, ctes: t.Dict[str, exp.Expr], recursive: bool = False) -> Non
701755
with self._concurrent_render_context():
702756
# Similar to the model's query, we render the CTE query under the locked context
703757
# so that the execution (fetchdf) can continue concurrently between the threads
704-
sql = cte_query.sql(
705-
self._test_adapter_dialect, pretty=self.engine_adapter._pretty_sql
706-
)
758+
sql = self._generate_sql(cte_query)
707759

708760
actual = self._execute(sql)
709761
expected = self._create_df(values, columns=cte_query.named_selects, partial=partial)
@@ -715,7 +767,7 @@ def runTest(self) -> None:
715767
# Render the model's query and generate the SQL under the locked context so that
716768
# execution (fetchdf) can continue concurrently between the threads
717769
query = self._render_model_query()
718-
sql = query.sql(self._test_adapter_dialect, pretty=self.engine_adapter._pretty_sql)
770+
sql = self._generate_sql(query)
719771

720772
with_clause = query.args.get("with_")
721773

@@ -820,7 +872,7 @@ def _execute_model(self) -> pd.DataFrame:
820872
"""Executes the python model and returns a DataFrame."""
821873
import pandas as pd
822874

823-
with self._concurrent_render_context():
875+
with self._concurrent_render_context(patch_shared_dialect=True):
824876
variables = self.body.get("vars", {}).copy()
825877
time_kwargs = {key: variables.pop(key) for key in TIME_KWARG_KEYS if key in variables}
826878
df = next(self.model.render(context=self.context, variables=variables, **time_kwargs))

tests/core/test_test.py

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
import datetime
44
import typing as t
55
import io
6+
import threading
67
from pathlib import Path
78
import unittest
89
from unittest.mock import call, patch
@@ -1360,6 +1361,68 @@ def test_nested_data_types(sushi_context: Context) -> None:
13601361
)
13611362

13621363

1364+
def test_freeze_time_does_not_patch_shared_generator() -> None:
1365+
from sqlglot.dialects.dialect import Dialect
1366+
from sqlglot.generator import _DISPATCH_CACHE
1367+
1368+
duckdb = Dialect.get_or_raise("duckdb")
1369+
exp.select("1").sql("duckdb") # make sure the shared dispatch table exists
1370+
shared_transforms = dict(duckdb.generator_class.TRANSFORMS)
1371+
shared_dispatch = dict(_DISPATCH_CACHE[duckdb.generator_class])
1372+
1373+
test = _create_test(
1374+
body=load_yaml(
1375+
"""
1376+
test_foo:
1377+
model: xyz
1378+
outputs:
1379+
query:
1380+
- cur_date: 2023-01-01
1381+
vars:
1382+
execution_time: "2023-01-01 12:05:03+00:00"
1383+
"""
1384+
),
1385+
test_name="test_foo",
1386+
model=_create_model("SELECT CURRENT_DATE AS cur_date"),
1387+
context=Context(config=Config(model_defaults=ModelDefaultsConfig(dialect="duckdb"))),
1388+
)
1389+
test.concurrency = True
1390+
1391+
rendered: t.List[str] = []
1392+
errors: t.List[BaseException] = []
1393+
1394+
def render_ddl_through_shared_dialect() -> None:
1395+
# Mimics another test creating its fixture views while this test's frozen render
1396+
# context is active; both go through the dialect's shared generator class.
1397+
try:
1398+
for _ in range(200):
1399+
rendered.append(
1400+
exp.Create(
1401+
this=exp.to_table("s.v"), kind="VIEW", expression=exp.select("1")
1402+
).sql("duckdb")
1403+
)
1404+
rendered.append(exp.CurrentDate().sql("duckdb"))
1405+
except BaseException as e: # pragma: no cover
1406+
errors.append(e)
1407+
1408+
with test._concurrent_render_context():
1409+
other = threading.Thread(target=render_ddl_through_shared_dialect)
1410+
other.start()
1411+
other.join()
1412+
1413+
# this test renders the frozen time...
1414+
assert test._generate_sql(exp.CurrentDate()) == "CAST('2023-01-01 12:05:03+00:00' AS DATE)"
1415+
# ...while the shared dialect is untouched, even inside the frozen context
1416+
assert exp.CurrentDate().sql("duckdb") == "CURRENT_DATE"
1417+
1418+
assert not errors
1419+
assert set(rendered) == {"CREATE VIEW s.v AS SELECT 1", "CURRENT_DATE"}
1420+
assert duckdb.generator_class.TRANSFORMS == shared_transforms
1421+
assert _DISPATCH_CACHE[duckdb.generator_class] == shared_dispatch
1422+
1423+
_check_successful_or_raise(test.run())
1424+
1425+
13631426
def test_freeze_time(mocker: MockerFixture) -> None:
13641427
mocker.patch("sqlmesh.core.test.definition.random_id", return_value="jzngz56a")
13651428
test = _create_test(

0 commit comments

Comments
 (0)