Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions sqlmesh/core/selector.py
Original file line number Diff line number Diff line change
Expand Up @@ -146,7 +146,27 @@ def get_model(fqn: str) -> t.Optional[Model]:
models[model.fqn] = model

if needs_update:
schema_update_fqns = set(dag.graph)
unselected_external_hashes = {
fqn: (model.json(), model.data_hash, model.metadata_hash)
for fqn, model in models.items()
if fqn in env_models
and fqn not in all_selected_models
and fqn in schema_update_fqns
and model.kind.is_external
}
update_model_schemas(dag, models=models, cache_dir=self._cache_dir)
for fqn, (
serialized_model,
data_hash,
metadata_hash,
) in unselected_external_hashes.items():
model = models.get(fqn)
if model and model.json() == serialized_model:
# Schema updates can rehash an unselected external model loaded from state.
# Keep its deployed hash when its serialized definition did not change.
model._data_hash = data_hash
model._metadata_hash = metadata_hash

return models, all_selected_models

Expand Down
159 changes: 157 additions & 2 deletions tests/core/test_selector_native.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,17 +9,172 @@
import subprocess

from sqlmesh.core import dialect as d
from sqlmesh.core.model import schema as schema_module
from sqlmesh.core import selector as selector_module
from sqlmesh.core.audit import StandaloneAudit
from sqlmesh.core.environment import Environment
from sqlmesh.core.model import Model, SqlModel
from sqlmesh.core.model import ExternalModel, Model, SqlModel, create_external_model
from sqlmesh.core.model.common import ParsableSql
from sqlmesh.core.selector import NativeSelector
from sqlmesh.core.snapshot import SnapshotChangeCategory
from sqlmesh.core.snapshot import Snapshot, SnapshotChangeCategory
from sqlmesh.core.snapshot.cache import SnapshotCache
from sqlmesh.core.snapshot.definition import Node
from sqlmesh.utils import UniqueKeyDict
from sqlmesh.utils.date import now_timestamp
from sqlmesh.utils.git import GitClient


def _external_schema_refresh_case(
mocker: MockerFixture, tmp_path: Path
) -> t.Tuple[
ExternalModel,
ExternalModel,
SqlModel,
t.Dict[str, Snapshot],
UniqueKeyDict[str, Model],
NativeSelector,
]:
external_a = create_external_model("db.external_a", columns={"id": "int"})
external_b = create_external_model("db.external_b", columns={"id": "int"})
child = SqlModel(
name="db.child",
query=d.parse_one(
"SELECT a.id FROM db.external_a AS a JOIN db.external_b AS b ON a.id = b.id"
),
)

# Inject inconsistent persisted state for this invariant test: the stored external model
# has optimize_query=False, but its fingerprint is seeded from the model with None. This
# synthetic mismatch does not imply that an upstream migration produces this state.
stored_b = t.cast(ExternalModel, external_b.copy(update={"optimize_query": False}))
stored_b._data_hash = external_b.data_hash
stored_b._metadata_hash = external_b.metadata_hash

old_nodes: t.Dict[str, Node] = {
external_a.fqn: external_a,
stored_b.fqn: stored_b,
child.fqn: child,
}
old_snapshots = {
name: Snapshot.from_node(node, nodes=old_nodes) for name, node in old_nodes.items()
}
for snapshot in old_snapshots.values():
snapshot.categorize_as(SnapshotChangeCategory.BREAKING)

current_models: UniqueKeyDict[str, Model] = UniqueKeyDict("models")
current_models[external_a.fqn] = external_a
current_models[external_b.fqn] = external_b
current_models[child.fqn] = child

# Simulate the process worker's fresh fingerprint calculation synchronously. This avoids
# forking from pytest while still exercising the schema worker's hash recomputation.
mocker.patch("sqlmesh.core.constants.MAX_FORK_WORKERS", 1)
original_load = schema_module.load_optimized_query_and_mapping

def load_with_fresh_hashes(model: Model, mapping: t.Dict) -> t.Any:
model._data_hash = None
model._metadata_hash = None
return original_load(model, mapping)

mocker.patch(
"sqlmesh.core.model.schema.load_optimized_query_and_mapping",
side_effect=load_with_fresh_hashes,
)

state_reader = mocker.Mock()
state_reader.get_environment.return_value = Environment(
name="prod",
snapshots=[snapshot.table_info for snapshot in old_snapshots.values()],
start_at="2023-01-01",
end_at="2023-01-02",
plan_id="test_plan",
)

snapshot_cache = SnapshotCache(tmp_path / "snapshot_cache")

def load_snapshots(infos):
snapshot_ids = {info.snapshot_id for info in infos}
snapshots, _ = snapshot_cache.get_or_load(
snapshot_ids,
lambda ids: [
snapshot.copy(deep=True)
for snapshot in old_snapshots.values()
if snapshot.snapshot_id in ids
],
)
return snapshots

state_reader.get_snapshots.side_effect = load_snapshots

selector = NativeSelector(
state_reader,
current_models,
context_path=tmp_path,
cache_dir=tmp_path,
)
return external_a, external_b, child, old_snapshots, current_models, selector


def _current_snapshots(models: UniqueKeyDict[str, Model]) -> t.Dict[str, Snapshot]:
nodes = t.cast(t.Dict[str, Node], dict(models))
return {model.fqn: Snapshot.from_node(model, nodes=nodes) for model in models.values()}


def test_unselected_external_coparent_is_not_directly_modified_by_schema_refresh(
mocker: MockerFixture, tmp_path: Path
) -> None:
external_a, external_b, child, old_snapshots, _, selector = _external_schema_refresh_case(
mocker, tmp_path
)

schema_refresh = mocker.spy(selector_module, "update_model_schemas")
models, selected_fqns = selector.select_models([external_a.fqn], "prod")
current = _current_snapshots(models)

assert external_a.fqn in selected_fqns
schema_refresh.assert_called_once()
assert not current[external_b.fqn].is_directly_modified(old_snapshots[external_b.fqn])
assert (
current[external_b.fqn].fingerprint.metadata_hash
== old_snapshots[external_b.fqn].fingerprint.metadata_hash
)
assert not current[child.fqn].is_indirectly_modified(old_snapshots[child.fqn])


def test_selected_external_change_remains_directly_modified(
mocker: MockerFixture, tmp_path: Path
) -> None:
_, external_b, _, old_snapshots, current_models, selector = _external_schema_refresh_case(
mocker, tmp_path
)
changed_b = create_external_model("db.external_b", columns={"id": "int", "new": "text"})
current_models.update({changed_b.fqn: changed_b})

models, selected_fqns = selector.select_models([external_b.fqn], "prod")
current = _current_snapshots(models)

assert external_b.fqn in selected_fqns
assert current[external_b.fqn].is_directly_modified(old_snapshots[external_b.fqn])


def test_selected_dependency_schema_change_remains_visible(
mocker: MockerFixture, tmp_path: Path
) -> None:
external_a, external_b, child, old_snapshots, current_models, selector = (
_external_schema_refresh_case(mocker, tmp_path)
)
changed_a = create_external_model("db.external_a", columns={"id": "int", "new": "text"})
current_models.update({changed_a.fqn: changed_a})

models, selected_fqns = selector.select_models([external_a.fqn], "prod")
current = _current_snapshots(models)

assert external_a.fqn in selected_fqns
assert current[external_a.fqn].is_directly_modified(old_snapshots[external_a.fqn])
assert not current[external_b.fqn].is_directly_modified(old_snapshots[external_b.fqn])
assert current[child.fqn].is_indirectly_modified(old_snapshots[child.fqn])


@pytest.mark.parametrize(
"default_catalog",
[
Expand Down
Loading