Skip to content
Merged
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
21 changes: 16 additions & 5 deletions sqlmesh/core/snapshot/evaluator.py
Original file line number Diff line number Diff line change
Expand Up @@ -1073,11 +1073,22 @@ def _cleanup_snapshot(

evaluation_strategy = _evaluation_strategy(snapshot, adapter)
for is_table_deployable, table_name in table_names:
evaluation_strategy.delete(
table_name,
is_table_deployable=is_table_deployable,
physical_schema=snapshot.physical_schema,
)
try:
evaluation_strategy.delete(
table_name,
is_table_deployable=is_table_deployable,
physical_schema=snapshot.physical_schema,
)
except Exception:
# Use `get_data_object` to check if the table exists instead of `table_exists` since the former
# is based on `INFORMATION_SCHEMA` and avoids touching the table directly.
# This is important when the table name is malformed for some reason and running any statement
# that touches the table would result in an error.
if adapter.get_data_object(table_name) is not None:
raise
logger.warning(
"Skipping cleanup of table '%s' because it does not exist", table_name
)

if on_complete is not None:
on_complete(table_name)
Expand Down
56 changes: 56 additions & 0 deletions tests/core/test_snapshot_evaluator.py
Original file line number Diff line number Diff line change
Expand Up @@ -438,12 +438,14 @@ def create_and_cleanup(name: str, dev_table_only: bool):
return snapshot

snapshot = create_and_cleanup("catalog.test_schema.test_model", True)
adapter_mock.get_data_object.assert_not_called()
adapter_mock.drop_table.assert_called_once_with(
f"catalog.sqlmesh__test_schema.test_schema__test_model__{snapshot.fingerprint.to_version()}__dev"
)
adapter_mock.reset_mock()

snapshot = create_and_cleanup("test_schema.test_model", False)
adapter_mock.get_data_object.assert_not_called()
adapter_mock.drop_table.assert_has_calls(
[
call(
Expand All @@ -455,6 +457,7 @@ def create_and_cleanup(name: str, dev_table_only: bool):
adapter_mock.reset_mock()

snapshot = create_and_cleanup("test_model", False)
adapter_mock.get_data_object.assert_not_called()
adapter_mock.drop_table.assert_has_calls(
[
call(f"sqlmesh__default.test_model__{snapshot.fingerprint.to_version()}__dev"),
Expand All @@ -463,6 +466,59 @@ def create_and_cleanup(name: str, dev_table_only: bool):
)


def test_cleanup_fails(adapter_mock, make_snapshot):
adapter_mock.drop_table.side_effect = RuntimeError("test_error")

evaluator = SnapshotEvaluator(adapter_mock)

model = SqlModel(
name="catalog.test_schema.test_model",
kind=IncrementalByTimeRangeKind(time_column="a"),
storage_format="parquet",
query=parse_one("SELECT a FROM tbl WHERE ds BETWEEN @start_ds and @end_ds"),
)

snapshot = make_snapshot(model)
snapshot.categorize_as(SnapshotChangeCategory.BREAKING, forward_only=True)
snapshot.version = "test_version"

evaluator.promote([snapshot], EnvironmentNamingInfo(name="test_env"))
with pytest.raises(NodeExecutionFailedError) as exc_info:
evaluator.cleanup(
[SnapshotTableCleanupTask(snapshot=snapshot.table_info, dev_table_only=True)]
)

assert str(exc_info.value.__cause__) == "test_error"


def test_cleanup_skip_missing_table(adapter_mock, make_snapshot):
adapter_mock.get_data_object.return_value = None
adapter_mock.drop_table.side_effect = RuntimeError("fail")

evaluator = SnapshotEvaluator(adapter_mock)

model = SqlModel(
name="catalog.test_schema.test_model",
kind=IncrementalByTimeRangeKind(time_column="a"),
storage_format="parquet",
query=parse_one("SELECT a FROM tbl WHERE ds BETWEEN @start_ds and @end_ds"),
)

snapshot = make_snapshot(model)
snapshot.categorize_as(SnapshotChangeCategory.BREAKING, forward_only=True)
snapshot.version = "test_version"

evaluator.promote([snapshot], EnvironmentNamingInfo(name="test_env"))
evaluator.cleanup([SnapshotTableCleanupTask(snapshot=snapshot.table_info, dev_table_only=True)])

adapter_mock.get_data_object.assert_called_once_with(
f"catalog.sqlmesh__test_schema.test_schema__test_model__{snapshot.fingerprint.to_version()}__dev"
)
adapter_mock.drop_table.assert_called_once_with(
f"catalog.sqlmesh__test_schema.test_schema__test_model__{snapshot.fingerprint.to_version()}__dev"
)


def test_cleanup_external_model(mocker: MockerFixture, adapter_mock, make_snapshot):
evaluator = SnapshotEvaluator(adapter_mock)

Expand Down