Fix: allow --select-model to plan a model deletion

Refactor select_models to return matched FQNs alongside the model dict
so context._plan can detect deletions without a separate env lookup.

Remove expand_model_selections_with_env, add mixed selection test coverage.

Fixes #5741

Signed-off-by: vinicius <viniciusnunes.tavares@gmail.com>

Author: viniciusnunest

sqlmesh/core/context.py

@@ -1605,13 +1605,11 @@ def plan_builder(
             backfill_models = None
 
         models_override: t.Optional[UniqueKeyDict[str, Model]] = None
-        # FQNs of models that are selected for deletion (present in the deployed environment but
-        # absent from local files). These are models whose files have been deleted; they will
-        # appear in context_diff.removed_snapshots rather than as backfill candidates.
+        selected_fqns: t.Set[str] = set()
         selected_deletion_fqns: t.Set[str] = set()
         if select_models:
             try:
-                models_override = model_selector.select_models(
+                models_override, selected_fqns = model_selector.select_models(
                     select_models,
                     environment,
                     fallback_env_name=create_from or c.PROD,
@@ -1627,16 +1625,9 @@ def plan_builder(
                 backfill_models = model_selector.expand_model_selections(select_models)
 
             if not backfill_models:
-                # The selection matched nothing locally. Check whether it matched models that exist
-                # in the deployed environment but have been deleted locally. If so, the selection is
-                # valid — the deletions will surface in context_diff.removed_snapshots.
-                env_selected = model_selector.expand_model_selections_with_env(
-                    select_models,
-                    environment,
-                    fallback_env_name=create_from or c.PROD,
-                    ensure_finalized_snapshots=self.config.plan.use_finalized_state,
-                )
-                selected_deletion_fqns = env_selected - set(self._models)
+                # The selection matched nothing locally. Check whether it matched models
+                # in the deployed environment that were deleted locally.
+                selected_deletion_fqns = selected_fqns - set(self._models)
 
         expanded_restate_models = None
         if restate_models is not None:

sqlmesh/core/selector.py

@@ -62,7 +62,7 @@ def select_models(
         target_env_name: str,
         fallback_env_name: t.Optional[str] = None,
         ensure_finalized_snapshots: bool = False,
-    ) -> UniqueKeyDict[str, Model]:
+    ) -> t.Tuple[UniqueKeyDict[str, Model], t.Set[str]]:
         """Given a set of selections returns models from the current state with names matching the
         selection while sourcing the remaining models from the target environment.
 
@@ -76,7 +76,7 @@ def select_models(
                 the environment is not finalized.
 
         Returns:
-            A dictionary of models.
+            A tuple of (models dict, set of all matched FQNs including env models).
         """
         env_models = self._load_env_models(
             target_env_name, fallback_env_name, ensure_finalized_snapshots
@@ -148,36 +148,7 @@ def get_model(fqn: str) -> t.Optional[Model]:
         if needs_update:
             update_model_schemas(dag, models=models, cache_dir=self._cache_dir)
 
-        return models
-
-    def expand_model_selections_with_env(
-        self,
-        model_selections: t.Iterable[str],
-        target_env_name: str,
-        fallback_env_name: t.Optional[str] = None,
-        ensure_finalized_snapshots: bool = False,
-    ) -> t.Set[str]:
-        """Expands model selections against both local models and the target environment.
-
-        This allows selections to match models that have been deleted locally but still
-        exist in the deployed environment.
-
-        Args:
-            model_selections: A set of selections.
-            target_env_name: The name of the target environment.
-            fallback_env_name: The name of the fallback environment that will be used if the target
-                environment doesn't exist.
-            ensure_finalized_snapshots: Whether to source environment snapshots from the latest finalized
-                environment state, or to use whatever snapshots are in the current environment state even if
-                the environment is not finalized.
-
-        Returns:
-            A set of matched model FQNs.
-        """
-        env_models = self._load_env_models(
-            target_env_name, fallback_env_name, ensure_finalized_snapshots
-        )
-        return self.expand_model_selections(model_selections, models={**env_models, **self._models})
+        return models, all_selected_models
 
     def _load_env_models(
         self,

tests/core/test_selector_native.py

@@ -309,7 +309,7 @@ def test_select_change_schema(mocker: MockerFixture, make_snapshot):
 
     selector = NativeSelector(state_reader_mock, local_models)
 
-    selected = selector.select_models(["db.parent"], env_name)
+    selected, _ = selector.select_models(["db.parent"], env_name)
     assert selected[local_child.fqn].render_query() != child.render_query()
 
     _assert_models_equal(
@@ -320,7 +320,7 @@ def test_select_change_schema(mocker: MockerFixture, make_snapshot):
         },
     )
 
-    selected = selector.select_models(["db.child"], env_name)
+    selected, _ = selector.select_models(["db.child"], env_name)
     assert selected[local_child.fqn].data_hash == child.data_hash
 
     _assert_models_equal(
@@ -343,12 +343,12 @@ def test_select_models_missing_env(mocker: MockerFixture, make_snapshot):
 
     selector = NativeSelector(state_reader_mock, local_models)
 
-    assert selector.select_models([model.name], "missing_env").keys() == {model.fqn}
-    assert not selector.select_models(["missing"], "missing_env")
+    assert selector.select_models([model.name], "missing_env")[0].keys() == {model.fqn}
+    assert not selector.select_models(["missing"], "missing_env")[0]
 
     assert selector.select_models(
         [model.name], "missing_env", fallback_env_name="another_missing_env"
-    ).keys() == {model.fqn}
+    )[0].keys() == {model.fqn}
 
     state_reader_mock.get_environment.assert_has_calls(
         [
@@ -789,7 +789,7 @@ def test_select_models_local_tags_take_precedence_over_remote(
 
     selector = NativeSelector(state_reader_mock, local_models)
 
-    selected = selector.select_models(["tag:a"], env_name)
+    selected, _ = selector.select_models(["tag:a"], env_name)
 
     # both should get selected because they both now have the 'a' tag locally, even though one exists in remote state without the 'a' tag
     _assert_models_equal(
@@ -801,9 +801,9 @@ def test_select_models_local_tags_take_precedence_over_remote(
     )
 
 
-def test_expand_model_selections_with_env(mocker: MockerFixture, make_snapshot):
-    """expand_model_selections_with_env should include models from the deployed environment,
-    even if they have been deleted locally."""
+def test_select_models_returns_selected_fqns(mocker: MockerFixture, make_snapshot):
+    """select_models should return the set of all matched FQNs (including env-only models)
+    alongside the model dict."""
     local_model = SqlModel(
         name="db.local_model",
         query=d.parse_one("SELECT 1 AS a"),
@@ -835,34 +835,31 @@ def test_expand_model_selections_with_env(mocker: MockerFixture, make_snapshot):
 
     selector = NativeSelector(state_reader_mock, local_models)
 
-    # Expanding against local models only should NOT find the deleted model.
-    assert selector.expand_model_selections(["db.deleted_model"]) == set()
+    # Selecting a deleted model: selected_fqns includes it even though models dict won't.
+    _, selected_fqns = selector.select_models(["db.deleted_model"], env_name)
+    assert deleted_model.fqn in selected_fqns
 
-    # Expanding with env should find the deleted model.
-    result = selector.expand_model_selections_with_env(["db.deleted_model"], env_name)
-    assert deleted_model.fqn in result
+    # Selecting a local model: selected_fqns includes it.
+    _, selected_fqns = selector.select_models(["db.local_model"], env_name)
+    assert local_model.fqn in selected_fqns
 
-    # Local model should also be reachable.
-    result = selector.expand_model_selections_with_env(["db.local_model"], env_name)
-    assert local_model.fqn in result
-
-    # Selecting both should return both.
-    result = selector.expand_model_selections_with_env(
+    # Mixed selection (active + deleted): both appear in selected_fqns.
+    _, selected_fqns = selector.select_models(
         ["db.deleted_model", "db.local_model"], env_name
     )
-    assert result == {deleted_model.fqn, local_model.fqn}
+    assert selected_fqns == {deleted_model.fqn, local_model.fqn}
 
-    # Wildcard should match env models too.
-    result = selector.expand_model_selections_with_env(["*_model"], env_name)
-    assert result == {deleted_model.fqn, local_model.fqn}
+    # Wildcard should match both local and env models.
+    _, selected_fqns = selector.select_models(["*_model"], env_name)
+    assert selected_fqns == {deleted_model.fqn, local_model.fqn}
 
-    # Non-existent model should return empty.
-    result = selector.expand_model_selections_with_env(["db.nonexistent"], env_name)
-    assert result == set()
+    # Non-existent model should not appear.
+    _, selected_fqns = selector.select_models(["db.nonexistent"], env_name)
+    assert selected_fqns == set()
 
 
-def test_expand_model_selections_with_env_fallback(mocker: MockerFixture, make_snapshot):
-    """expand_model_selections_with_env should fall back to the fallback environment."""
+def test_select_models_selected_fqns_fallback(mocker: MockerFixture, make_snapshot):
+    """select_models selected_fqns should include env models found via fallback environment."""
     deleted_model = SqlModel(
         name="db.deleted_model",
         query=d.parse_one("SELECT 1 AS a"),
@@ -890,14 +887,14 @@ def test_expand_model_selections_with_env_fallback(mocker: MockerFixture, make_s
     local_models: UniqueKeyDict[str, Model] = UniqueKeyDict("models")
     selector = NativeSelector(state_reader_mock, local_models)
 
-    result = selector.expand_model_selections_with_env(
+    _, selected_fqns = selector.select_models(
         ["db.deleted_model"], "missing_env", fallback_env_name="prod"
     )
-    assert deleted_model.fqn in result
+    assert deleted_model.fqn in selected_fqns
 
 
-def test_expand_model_selections_with_env_expired(mocker: MockerFixture, make_snapshot):
-    """expand_model_selections_with_env should ignore expired environments."""
+def test_select_models_selected_fqns_expired(mocker: MockerFixture, make_snapshot):
+    """select_models should not match env models from expired environments."""
     deleted_model = SqlModel(
         name="db.deleted_model",
         query=d.parse_one("SELECT 1 AS a"),
@@ -924,11 +921,17 @@ def test_expand_model_selections_with_env_expired(mocker: MockerFixture, make_sn
     local_models: UniqueKeyDict[str, Model] = UniqueKeyDict("models")
     selector = NativeSelector(state_reader_mock, local_models)
 
-    result = selector.expand_model_selections_with_env(["db.deleted_model"], "test_env")
-    assert result == set()
+    _, selected_fqns = selector.select_models(["db.deleted_model"], "test_env")
+    assert selected_fqns == set()
 
 
-def _assert_models_equal(actual: t.Dict[str, Model], expected: t.Dict[str, Model]) -> None:
+def _assert_models_equal(
+    actual: t.Union[t.Dict[str, Model], t.Tuple[t.Dict[str, Model], t.Set[str]]],
+    expected: t.Dict[str, Model],
+) -> None:
+    # select_models returns a tuple; unwrap if needed.
+    if isinstance(actual, tuple):
+        actual = actual[0]
     assert set(actual) == set(expected)
     for name, model in actual.items():
         # Use dict() to make Pydantic V2 happy.