From 7597dbd02322cdbe028d34511251d6e6fbc66539 Mon Sep 17 00:00:00 2001
From: Erin Drummond <erin.dru@gmail.com>
Date: Tue, 26 Aug 2025 04:44:31 +0000
Subject: [PATCH 1/7] Feat(dbt_cli): Set proper plan flags and also allow
 --empty and --environment

---
 sqlmesh/core/context.py          |  10 +-
 sqlmesh_dbt/cli.py               |   9 ++
 sqlmesh_dbt/operations.py        | 106 ++++++++++++++++++---
 tests/dbt/cli/test_operations.py | 155 +++++++++++++++++++++++++++++++
 4 files changed, 266 insertions(+), 14 deletions(-)

diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py
index 1a5375183c..59d1d4ab43 100644
--- a/sqlmesh/core/context.py
+++ b/sqlmesh/core/context.py
@@ -1296,6 +1296,7 @@ def plan(
         explain: t.Optional[bool] = None,
         ignore_cron: t.Optional[bool] = None,
         min_intervals: t.Optional[int] = None,
+        always_recreate_environment: t.Optional[bool] = None,
     ) -> Plan:
         """Interactively creates a plan.
 
@@ -1345,6 +1346,7 @@ def plan(
             explain: Whether to explain the plan instead of applying it.
             min_intervals: Adjust the plan start date on a per-model basis in order to ensure at least this many intervals are covered
                 on every model when checking for missing intervals
+            always_recreate_environment: Whether to always recreate the target environment from the `create_from` environment.
 
         Returns:
             The populated Plan object.
@@ -1376,6 +1378,7 @@ def plan(
             explain=explain,
             ignore_cron=ignore_cron,
             min_intervals=min_intervals,
+            always_recreate_environment=always_recreate_environment,
         )
 
         plan = plan_builder.build()
@@ -1428,6 +1431,7 @@ def plan_builder(
         explain: t.Optional[bool] = None,
         ignore_cron: t.Optional[bool] = None,
         min_intervals: t.Optional[int] = None,
+        always_recreate_environment: t.Optional[bool] = None,
     ) -> PlanBuilder:
         """Creates a plan builder.
 
@@ -1466,7 +1470,7 @@ def plan_builder(
             diff_rendered: Whether the diff should compare raw vs rendered models
             min_intervals: Adjust the plan start date on a per-model basis in order to ensure at least this many intervals are covered
                 on every model when checking for missing intervals
-
+            always_recreate_environment: Whether to always recreate the target environment from the `create_from` environment.
         Returns:
             The plan builder.
         """
@@ -1497,6 +1501,7 @@ def plan_builder(
             "diff_rendered": diff_rendered,
             "skip_linter": skip_linter,
             "min_intervals": min_intervals,
+            "always_recreate_environment": always_recreate_environment,
         }
         user_provided_flags: t.Dict[str, UserProvidedFlags] = {
             k: v for k, v in kwargs.items() if v is not None
@@ -1588,7 +1593,8 @@ def plan_builder(
             or (backfill_models is not None and not backfill_models),
             ensure_finalized_snapshots=self.config.plan.use_finalized_state,
             diff_rendered=diff_rendered,
-            always_recreate_environment=self.config.plan.always_recreate_environment,
+            always_recreate_environment=always_recreate_environment
+            or self.config.plan.always_recreate_environment,
         )
         modified_model_names = {
             *context_diff.modified_snapshots,
diff --git a/sqlmesh_dbt/cli.py b/sqlmesh_dbt/cli.py
index c215663f0a..6b724f8b7f 100644
--- a/sqlmesh_dbt/cli.py
+++ b/sqlmesh_dbt/cli.py
@@ -92,7 +92,15 @@ def dbt(
     "--full-refresh",
     help="If specified, dbt will drop incremental models and fully-recalculate the incremental table from the model definition.",
 )
+@click.option(
+    "--environment",
+    help="Run against a specific Virtual Data Environment (VDE) instead of the main environment",
+)
+@click.option(
+    "--empty/--no-empty", default=False, help="If specified, limit input refs and sources"
+)
 @vars_option
+@cli_global_error_handler
 @click.pass_context
 def run(ctx: click.Context, vars: t.Optional[t.Dict[str, t.Any]], **kwargs: t.Any) -> None:
     """Compile SQL and execute against the current target database."""
@@ -103,6 +111,7 @@ def run(ctx: click.Context, vars: t.Optional[t.Dict[str, t.Any]], **kwargs: t.An
 @select_option
 @exclude_option
 @vars_option
+@cli_global_error_handler
 @click.pass_context
 def list_(ctx: click.Context, vars: t.Optional[t.Dict[str, t.Any]], **kwargs: t.Any) -> None:
     """List the resources in your project"""
diff --git a/sqlmesh_dbt/operations.py b/sqlmesh_dbt/operations.py
index 270bba6511..9cf1968bb6 100644
--- a/sqlmesh_dbt/operations.py
+++ b/sqlmesh_dbt/operations.py
@@ -11,6 +11,7 @@
     from sqlmesh.dbt.project import Project
     from sqlmesh_dbt.console import DbtCliConsole
     from sqlmesh.core.model import Model
+    from sqlmesh.core.plan import Plan
 
 logger = logging.getLogger(__name__)
 
@@ -35,21 +36,20 @@ def list_(
 
     def run(
         self,
+        environment: t.Optional[str] = None,
         select: t.Optional[t.List[str]] = None,
         exclude: t.Optional[t.List[str]] = None,
         full_refresh: bool = False,
-    ) -> None:
-        select_models = None
-
-        if sqlmesh_selector := selectors.to_sqlmesh(select or [], exclude or []):
-            select_models = [sqlmesh_selector]
-
-        self.context.plan(
-            select_models=select_models,
-            run=True,
-            no_diff=True,
-            no_prompts=True,
-            auto_apply=True,
+        empty: bool = False,
+    ) -> Plan:
+        return self.context.plan(
+            **self._plan_options(
+                environment=environment,
+                select=select,
+                exclude=exclude,
+                full_refresh=full_refresh,
+                empty=empty,
+            )
         )
 
     def _selected_models(
@@ -71,6 +71,88 @@ def _selected_models(
 
         return selected_models
 
+    def _plan_options(
+        self,
+        environment: t.Optional[str] = None,
+        select: t.Optional[t.List[str]] = None,
+        exclude: t.Optional[t.List[str]] = None,
+        empty: bool = False,
+        full_refresh: bool = False,
+    ) -> t.Dict[str, t.Any]:
+        import sqlmesh.core.constants as c
+
+        # convert --select and --exclude to a selector expression for the SQLMesh selector engine
+        select_models = None
+        if sqlmesh_selector := selectors.to_sqlmesh(select or [], exclude or []):
+            select_models = [sqlmesh_selector]
+
+        is_dev = environment and environment != c.PROD
+        is_prod = not is_dev
+
+        options: t.Dict[str, t.Any] = {}
+
+        if is_prod or (is_dev and select_models):
+            # prod plans should "catch up" before applying the changes so that after the command finishes prod is the latest it can be
+            # dev plans *with* selectors should do the same as the user is saying "specifically update these models to the latest"
+            # dev plans *without* selectors should just have the defaults of never exceeding prod as the user is saying "just create this env" without focusing on any specific models
+            options.update(
+                dict(
+                    # always catch the data up to latest rather than only operating on what has been loaded before
+                    run=True,
+                    # don't taking cron schedules into account when deciding what models to run, do everything even if it just ran
+                    ignore_cron=True,
+                )
+            )
+
+        if is_dev:
+            options.update(
+                dict(
+                    # don't create views for all of prod in the dev environment
+                    include_unmodified=False,
+                    # always plan from scratch against prod rather than planning against the previous state of an existing dev environment
+                    # this results in the full scope of changes vs prod always being shown on the local branch
+                    create_from=c.PROD,
+                    always_recreate_environment=True,
+                    # setting enable_preview=None enables dev previews of forward_only changes for dbt projects IF the target engine supports cloning
+                    # if we set enable_preview=True here, this enables dev previews in all cases.
+                    # In the case of dbt default INCREMENTAL_UNMANAGED models, this will cause incremental models to be fully rebuilt (potentially a very large computation)
+                    # just to have the results thrown away on promotion to prod because dev previews are not promotable.
+                    #
+                    # TODO: if the user "upgrades" to an INCREMENTAL_BY_TIME_RANGE by defining a "time_column", we can inject leak guards to compute
+                    # just a preview instead of the whole thing like we would in a native project, but the enable_preview setting is at the plan level
+                    # and not the individual model level so we currently have no way of doing this selectively
+                    enable_preview=None,
+                )
+            )
+
+        if empty:
+            # dbt --empty adds LIMIT 0 to the queries, resulting in empty tables
+            # this lines up with --skip-backfill in SQLMesh
+            options["skip_backfill"] = True
+
+            if is_prod:
+                # to prevent the following error:
+                # > ConfigError: When targeting the production environment either the backfill should not be skipped or
+                # > the lack of data gaps should be enforced (--no-gaps flag).
+                options["no_gaps"] = True
+
+        if full_refresh:
+            # TODO: handling this requires some updates in the engine to enable restatements+changes in the same plan without affecting prod
+            # if the plan targets dev
+            pass
+
+        return dict(
+            environment=environment,
+            select_models=select_models,
+            # dont output a diff of model changes
+            no_diff=True,
+            # don't throw up any prompts like "set the effective date" - use defaults
+            no_prompts=True,
+            # start doing work immediately (since no_diff is set, there isnt really anything for the user to say yes/no to)
+            auto_apply=True,
+            **options,
+        )
+
     @property
     def console(self) -> DbtCliConsole:
         console = self.context.console
diff --git a/tests/dbt/cli/test_operations.py b/tests/dbt/cli/test_operations.py
index 9b5b3113b3..56ea23941c 100644
--- a/tests/dbt/cli/test_operations.py
+++ b/tests/dbt/cli/test_operations.py
@@ -1,13 +1,35 @@
+import typing as t
 from pathlib import Path
 import pytest
 from sqlmesh_dbt.operations import create
 from sqlmesh.utils import yaml
 from sqlmesh.utils.errors import SQLMeshError
 import time_machine
+from sqlmesh.core.console import NoopConsole
+from sqlmesh.core.plan import PlanBuilder
 
 pytestmark = pytest.mark.slow
 
 
+class PlanCapturingConsole(NoopConsole):
+    def plan(
+        self,
+        plan_builder: PlanBuilder,
+        auto_apply: bool,
+        default_catalog: t.Optional[str],
+        no_diff: bool = False,
+        no_prompts: bool = False,
+    ) -> None:
+        self.plan_builder = plan_builder
+        self.auto_apply = auto_apply
+        self.default_catalog = default_catalog
+        self.no_diff = no_diff
+        self.no_prompts = no_prompts
+
+        # normal console starts applying the plan here; we dont because we just want to capture the parameters
+        # and check they were set correctly
+
+
 def test_create_sets_and_persists_default_start_date(jaffle_shop_duckdb: Path):
     with time_machine.travel("2020-01-02 00:00:00 UTC"):
         from sqlmesh.utils.date import yesterday_ds, to_ds
@@ -83,3 +105,136 @@ def test_create_can_set_project_variables(jaffle_shop_duckdb: Path):
     query = test_model.render_query()
     assert query is not None
     assert query.sql() == "SELECT 'bar' AS \"a\""
+
+
+def test_run_option_mapping(jaffle_shop_duckdb: Path):
+    operations = create(project_dir=jaffle_shop_duckdb)
+    console = PlanCapturingConsole()
+    operations.context.console = console
+
+    plan = operations.run()
+    assert plan.environment.name == "prod"
+    assert console.no_prompts is True
+    assert console.no_diff is True
+    assert console.auto_apply is True
+    assert plan.end_bounded is False
+    assert plan.ignore_cron is True
+    assert plan.skip_backfill is False
+    assert plan.selected_models_to_backfill is None
+    assert {s.name for s in plan.snapshots} == {k for k in operations.context.snapshots}
+
+    plan = operations.run(select=["main.stg_orders+"])
+    assert plan.environment.name == "prod"
+    assert console.no_prompts is True
+    assert console.no_diff is True
+    assert console.auto_apply is True
+    assert plan.end_bounded is False
+    assert plan.ignore_cron is True
+    assert plan.skip_backfill is False
+    assert plan.selected_models_to_backfill == {
+        '"jaffle_shop"."main"."customers"',
+        '"jaffle_shop"."main"."orders"',
+        '"jaffle_shop"."main"."stg_orders"',
+    }
+    assert {s.name for s in plan.snapshots} == plan.selected_models_to_backfill
+
+    plan = operations.run(select=["main.stg_orders+"], exclude=["main.customers"])
+    assert plan.environment.name == "prod"
+    assert console.no_prompts is True
+    assert console.no_diff is True
+    assert console.auto_apply is True
+    assert plan.end_bounded is False
+    assert plan.ignore_cron is True
+    assert plan.skip_backfill is False
+    assert plan.selected_models_to_backfill == {
+        '"jaffle_shop"."main"."orders"',
+        '"jaffle_shop"."main"."stg_orders"',
+    }
+    assert {s.name for s in plan.snapshots} == plan.selected_models_to_backfill
+
+    plan = operations.run(exclude=["main.customers"])
+    assert plan.environment.name == "prod"
+    assert console.no_prompts is True
+    assert console.no_diff is True
+    assert console.auto_apply is True
+    assert plan.end_bounded is False
+    assert plan.ignore_cron is True
+    assert plan.skip_backfill is False
+    assert plan.selected_models_to_backfill == {k for k in operations.context.snapshots} - {
+        '"jaffle_shop"."main"."customers"'
+    }
+    assert {s.name for s in plan.snapshots} == plan.selected_models_to_backfill
+
+    plan = operations.run(empty=True)
+    assert plan.environment.name == "prod"
+    assert console.no_prompts is True
+    assert console.no_diff is True
+    assert console.auto_apply is True
+    assert plan.end_bounded is False
+    assert plan.ignore_cron is True
+    assert plan.skip_backfill is True
+    assert plan.selected_models_to_backfill is None
+    assert {s.name for s in plan.snapshots} == {k for k in operations.context.snapshots}
+
+
+def test_run_option_mapping_dev(jaffle_shop_duckdb: Path):
+    # create prod so that dev has something to compare against
+    operations = create(project_dir=jaffle_shop_duckdb)
+    operations.run()
+
+    (jaffle_shop_duckdb / "models" / "new_model.sql").write_text("select 1")
+
+    operations = create(project_dir=jaffle_shop_duckdb)
+
+    console = PlanCapturingConsole()
+    operations.context.console = console
+
+    plan = operations.run(environment="dev")
+    assert plan.environment.name == "dev"
+    assert console.no_prompts is True
+    assert console.no_diff is True
+    assert console.auto_apply is True
+    assert plan.include_unmodified is False
+    assert plan.context_diff.create_from == "prod"
+    assert plan.context_diff.is_new_environment is True
+    assert (
+        console.plan_builder._enable_preview is False
+    )  # duckdb doesnt support cloning so dev previews are not enabled for dbt projects
+    assert plan.end_bounded is True
+    assert plan.ignore_cron is False
+    assert plan.skip_backfill is False
+    assert plan.selected_models_to_backfill == {'"jaffle_shop"."main"."new_model"'}
+
+    plan = operations.run(environment="dev", empty=True)
+    assert plan.environment.name == "dev"
+    assert console.no_prompts is True
+    assert console.no_diff is True
+    assert console.auto_apply is True
+    assert plan.include_unmodified is False
+    assert plan.context_diff.create_from == "prod"
+    assert plan.context_diff.is_new_environment is True
+    assert console.plan_builder._enable_preview is False
+    assert plan.end_bounded is True
+    assert plan.ignore_cron is False
+    assert plan.skip_backfill is True
+    assert plan.selected_models_to_backfill == {'"jaffle_shop"."main"."new_model"'}
+
+    plan = operations.run(environment="dev", select=["main.stg_orders+"])
+    assert plan.environment.name == "dev"
+    assert console.no_prompts is True
+    assert console.no_diff is True
+    assert console.auto_apply is True
+    assert plan.include_unmodified is False
+    assert plan.context_diff.create_from == "prod"
+    assert plan.context_diff.is_new_environment is True
+    assert console.plan_builder._enable_preview is False
+    # dev plans with --select have run=True, ignore_cron=True set
+    assert plan.end_bounded is False
+    assert plan.ignore_cron is True
+    assert plan.skip_backfill is False
+    # note: the new model in the dev environment is ignored in favour of the explicitly selected ones
+    assert plan.selected_models_to_backfill == {
+        '"jaffle_shop"."main"."customers"',
+        '"jaffle_shop"."main"."orders"',
+        '"jaffle_shop"."main"."stg_orders"',
+    }

From 7ddec9d3bd9456626d8e8cd6ba896cffcfa1ed2f Mon Sep 17 00:00:00 2001
From: Erin Drummond <erin.dru@gmail.com>
Date: Tue, 26 Aug 2025 20:58:35 +0000
Subject: [PATCH 2/7] Clarify enable_preview setting

---
 sqlmesh_dbt/operations.py        | 18 ++++++++++--------
 tests/dbt/cli/test_operations.py |  9 ++++-----
 2 files changed, 14 insertions(+), 13 deletions(-)

diff --git a/sqlmesh_dbt/operations.py b/sqlmesh_dbt/operations.py
index 9cf1968bb6..5dbb4050a1 100644
--- a/sqlmesh_dbt/operations.py
+++ b/sqlmesh_dbt/operations.py
@@ -113,15 +113,17 @@ def _plan_options(
                     # this results in the full scope of changes vs prod always being shown on the local branch
                     create_from=c.PROD,
                     always_recreate_environment=True,
-                    # setting enable_preview=None enables dev previews of forward_only changes for dbt projects IF the target engine supports cloning
-                    # if we set enable_preview=True here, this enables dev previews in all cases.
-                    # In the case of dbt default INCREMENTAL_UNMANAGED models, this will cause incremental models to be fully rebuilt (potentially a very large computation)
-                    # just to have the results thrown away on promotion to prod because dev previews are not promotable.
+                    # Always enable dev previews for incremental / forward-only models.
+                    # Due to how DBT does incrementals (INCREMENTAL_UNMANAGED on the SQLMesh engine), this will result in the full model being refreshed
+                    # with the entire dataset, which can potentially be very large. If this is undesirable, users have two options:
+                    #  - work around this using jinja to conditionally add extra filters to the WHERE clause or a LIMIT to the model query
+                    #  - upgrade to SQLMesh's incremental models, where we have variables for the start/end date and inject leak guards to
+                    #    limit the amount of data backfilled
                     #
-                    # TODO: if the user "upgrades" to an INCREMENTAL_BY_TIME_RANGE by defining a "time_column", we can inject leak guards to compute
-                    # just a preview instead of the whole thing like we would in a native project, but the enable_preview setting is at the plan level
-                    # and not the individual model level so we currently have no way of doing this selectively
-                    enable_preview=None,
+                    # Note: enable_preview=True is *different* behaviour to the `sqlmesh` CLI, which uses enable_preview=None.
+                    # This means the `sqlmesh` CLI will only enable dev previews for dbt projects if the target adapter supports cloning,
+                    # whereas we enable it unconditionally here
+                    enable_preview=True,
                 )
             )
 
diff --git a/tests/dbt/cli/test_operations.py b/tests/dbt/cli/test_operations.py
index 56ea23941c..0b0f8da9f4 100644
--- a/tests/dbt/cli/test_operations.py
+++ b/tests/dbt/cli/test_operations.py
@@ -197,9 +197,7 @@ def test_run_option_mapping_dev(jaffle_shop_duckdb: Path):
     assert plan.include_unmodified is False
     assert plan.context_diff.create_from == "prod"
     assert plan.context_diff.is_new_environment is True
-    assert (
-        console.plan_builder._enable_preview is False
-    )  # duckdb doesnt support cloning so dev previews are not enabled for dbt projects
+    assert console.plan_builder._enable_preview is True
     assert plan.end_bounded is True
     assert plan.ignore_cron is False
     assert plan.skip_backfill is False
@@ -213,7 +211,7 @@ def test_run_option_mapping_dev(jaffle_shop_duckdb: Path):
     assert plan.include_unmodified is False
     assert plan.context_diff.create_from == "prod"
     assert plan.context_diff.is_new_environment is True
-    assert console.plan_builder._enable_preview is False
+    assert console.plan_builder._enable_preview is True
     assert plan.end_bounded is True
     assert plan.ignore_cron is False
     assert plan.skip_backfill is True
@@ -227,8 +225,9 @@ def test_run_option_mapping_dev(jaffle_shop_duckdb: Path):
     assert plan.include_unmodified is False
     assert plan.context_diff.create_from == "prod"
     assert plan.context_diff.is_new_environment is True
-    assert console.plan_builder._enable_preview is False
+    assert console.plan_builder._enable_preview is True
     # dev plans with --select have run=True, ignore_cron=True set
+    # as opposed to dev plans that dont have a specific selector
     assert plan.end_bounded is False
     assert plan.ignore_cron is True
     assert plan.skip_backfill is False

From 26999e13fe9d5cca97381c92c7b98e6f014e8cb6 Mon Sep 17 00:00:00 2001
From: Erin Drummond <erin.dru@gmail.com>
Date: Wed, 27 Aug 2025 20:48:47 +0000
Subject: [PATCH 3/7] PR feedback

---
 sqlmesh/cli/project_init.py      |  7 +++++++
 sqlmesh/core/context.py          |  9 +--------
 sqlmesh_dbt/cli.py               |  1 +
 sqlmesh_dbt/operations.py        |  5 ++---
 tests/dbt/cli/test_operations.py | 13 +++++++++++++
 5 files changed, 24 insertions(+), 11 deletions(-)

diff --git a/sqlmesh/cli/project_init.py b/sqlmesh/cli/project_init.py
index 87d77da4b9..81ff534dc4 100644
--- a/sqlmesh/cli/project_init.py
+++ b/sqlmesh/cli/project_init.py
@@ -121,6 +121,13 @@ def _gen_config(
 # https://sqlmesh.readthedocs.io/en/stable/guides/configuration/#virtual-data-environment-modes
 virtual_environment_mode: {VirtualEnvironmentMode.DEV_ONLY.lower()}
 
+# --- Plan Defaults ---
+# https://sqlmesh.readthedocs.io/en/stable/reference/configuration/#plan
+plan:
+  # For Virtual Data Environments, this ensures that any changes are always considered against prod,
+  # rather than the previous state of that environment
+  always_recreate_environment: True
+
 # --- Model Defaults ---
 # https://sqlmesh.readthedocs.io/en/stable/reference/model_configuration/#model-defaults
 model_defaults:
diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py
index 59d1d4ab43..f877670914 100644
--- a/sqlmesh/core/context.py
+++ b/sqlmesh/core/context.py
@@ -1296,7 +1296,6 @@ def plan(
         explain: t.Optional[bool] = None,
         ignore_cron: t.Optional[bool] = None,
         min_intervals: t.Optional[int] = None,
-        always_recreate_environment: t.Optional[bool] = None,
     ) -> Plan:
         """Interactively creates a plan.
 
@@ -1346,7 +1345,6 @@ def plan(
             explain: Whether to explain the plan instead of applying it.
             min_intervals: Adjust the plan start date on a per-model basis in order to ensure at least this many intervals are covered
                 on every model when checking for missing intervals
-            always_recreate_environment: Whether to always recreate the target environment from the `create_from` environment.
 
         Returns:
             The populated Plan object.
@@ -1378,7 +1376,6 @@ def plan(
             explain=explain,
             ignore_cron=ignore_cron,
             min_intervals=min_intervals,
-            always_recreate_environment=always_recreate_environment,
         )
 
         plan = plan_builder.build()
@@ -1431,7 +1428,6 @@ def plan_builder(
         explain: t.Optional[bool] = None,
         ignore_cron: t.Optional[bool] = None,
         min_intervals: t.Optional[int] = None,
-        always_recreate_environment: t.Optional[bool] = None,
     ) -> PlanBuilder:
         """Creates a plan builder.
 
@@ -1470,7 +1466,6 @@ def plan_builder(
             diff_rendered: Whether the diff should compare raw vs rendered models
             min_intervals: Adjust the plan start date on a per-model basis in order to ensure at least this many intervals are covered
                 on every model when checking for missing intervals
-            always_recreate_environment: Whether to always recreate the target environment from the `create_from` environment.
         Returns:
             The plan builder.
         """
@@ -1501,7 +1496,6 @@ def plan_builder(
             "diff_rendered": diff_rendered,
             "skip_linter": skip_linter,
             "min_intervals": min_intervals,
-            "always_recreate_environment": always_recreate_environment,
         }
         user_provided_flags: t.Dict[str, UserProvidedFlags] = {
             k: v for k, v in kwargs.items() if v is not None
@@ -1593,8 +1587,7 @@ def plan_builder(
             or (backfill_models is not None and not backfill_models),
             ensure_finalized_snapshots=self.config.plan.use_finalized_state,
             diff_rendered=diff_rendered,
-            always_recreate_environment=always_recreate_environment
-            or self.config.plan.always_recreate_environment,
+            always_recreate_environment=self.config.plan.always_recreate_environment,
         )
         modified_model_names = {
             *context_diff.modified_snapshots,
diff --git a/sqlmesh_dbt/cli.py b/sqlmesh_dbt/cli.py
index 6b724f8b7f..c0a81d2c73 100644
--- a/sqlmesh_dbt/cli.py
+++ b/sqlmesh_dbt/cli.py
@@ -93,6 +93,7 @@ def dbt(
     help="If specified, dbt will drop incremental models and fully-recalculate the incremental table from the model definition.",
 )
 @click.option(
+    "--env",
     "--environment",
     help="Run against a specific Virtual Data Environment (VDE) instead of the main environment",
 )
diff --git a/sqlmesh_dbt/operations.py b/sqlmesh_dbt/operations.py
index 5dbb4050a1..beec6af068 100644
--- a/sqlmesh_dbt/operations.py
+++ b/sqlmesh_dbt/operations.py
@@ -109,10 +109,9 @@ def _plan_options(
                 dict(
                     # don't create views for all of prod in the dev environment
                     include_unmodified=False,
-                    # always plan from scratch against prod rather than planning against the previous state of an existing dev environment
-                    # this results in the full scope of changes vs prod always being shown on the local branch
+                    # always plan from scratch against prod. note that this is coupled with the `always_recreate_environment=True` setting in the default config file.
+                    # the result is that rather than planning against the previous state of an existing dev environment, the full scope of changes vs prod are always shown
                     create_from=c.PROD,
-                    always_recreate_environment=True,
                     # Always enable dev previews for incremental / forward-only models.
                     # Due to how DBT does incrementals (INCREMENTAL_UNMANAGED on the SQLMesh engine), this will result in the full model being refreshed
                     # with the entire dataset, which can potentially be very large. If this is undesirable, users have two options:
diff --git a/tests/dbt/cli/test_operations.py b/tests/dbt/cli/test_operations.py
index 0b0f8da9f4..15051542c6 100644
--- a/tests/dbt/cli/test_operations.py
+++ b/tests/dbt/cli/test_operations.py
@@ -7,6 +7,7 @@
 import time_machine
 from sqlmesh.core.console import NoopConsole
 from sqlmesh.core.plan import PlanBuilder
+from sqlmesh.core.config.common import VirtualEnvironmentMode
 
 pytestmark = pytest.mark.slow
 
@@ -93,6 +94,18 @@ def test_create_can_specify_profile_and_target(jaffle_shop_duckdb: Path):
     assert dbt_project.context.target_name == "dev"
 
 
+def test_default_options(jaffle_shop_duckdb: Path):
+    operations = create()
+
+    config = operations.context.config
+    dbt_project = operations.project
+
+    assert config.plan.always_recreate_environment is True
+    assert config.virtual_environment_mode == VirtualEnvironmentMode.DEV_ONLY
+    assert config.model_defaults.start is not None
+    assert config.model_defaults.dialect == dbt_project.context.target.dialect
+
+
 def test_create_can_set_project_variables(jaffle_shop_duckdb: Path):
     (jaffle_shop_duckdb / "models" / "test_model.sql").write_text("""
        select '{{ var('foo') }}' as a

From c2b6a03ef56512eae54c7e121b29ee13c4bffdcd Mon Sep 17 00:00:00 2001
From: Erin Drummond <erin.dru@gmail.com>
Date: Wed, 27 Aug 2025 20:55:38 +0000
Subject: [PATCH 4/7] tidyup

---
 sqlmesh/core/context.py | 1 +
 sqlmesh_dbt/cli.py      | 2 --
 2 files changed, 1 insertion(+), 2 deletions(-)

diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py
index f877670914..1a5375183c 100644
--- a/sqlmesh/core/context.py
+++ b/sqlmesh/core/context.py
@@ -1466,6 +1466,7 @@ def plan_builder(
             diff_rendered: Whether the diff should compare raw vs rendered models
             min_intervals: Adjust the plan start date on a per-model basis in order to ensure at least this many intervals are covered
                 on every model when checking for missing intervals
+
         Returns:
             The plan builder.
         """
diff --git a/sqlmesh_dbt/cli.py b/sqlmesh_dbt/cli.py
index c0a81d2c73..3b058abd17 100644
--- a/sqlmesh_dbt/cli.py
+++ b/sqlmesh_dbt/cli.py
@@ -101,7 +101,6 @@ def dbt(
     "--empty/--no-empty", default=False, help="If specified, limit input refs and sources"
 )
 @vars_option
-@cli_global_error_handler
 @click.pass_context
 def run(ctx: click.Context, vars: t.Optional[t.Dict[str, t.Any]], **kwargs: t.Any) -> None:
     """Compile SQL and execute against the current target database."""
@@ -112,7 +111,6 @@ def run(ctx: click.Context, vars: t.Optional[t.Dict[str, t.Any]], **kwargs: t.An
 @select_option
 @exclude_option
 @vars_option
-@cli_global_error_handler
 @click.pass_context
 def list_(ctx: click.Context, vars: t.Optional[t.Dict[str, t.Any]], **kwargs: t.Any) -> None:
     """List the resources in your project"""

From 83cfc2ba06b13f401960cc5f2d3a1bd7fcb22db2 Mon Sep 17 00:00:00 2001
From: Erin Drummond <erin.dru@gmail.com>
Date: Wed, 27 Aug 2025 21:18:39 +0000
Subject: [PATCH 5/7] Fix tests

---
 sqlmesh_dbt/cli.py | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/sqlmesh_dbt/cli.py b/sqlmesh_dbt/cli.py
index 3b058abd17..370f115d61 100644
--- a/sqlmesh_dbt/cli.py
+++ b/sqlmesh_dbt/cli.py
@@ -102,9 +102,14 @@ def dbt(
 )
 @vars_option
 @click.pass_context
-def run(ctx: click.Context, vars: t.Optional[t.Dict[str, t.Any]], **kwargs: t.Any) -> None:
+def run(
+    ctx: click.Context,
+    vars: t.Optional[t.Dict[str, t.Any]],
+    env: t.Optional[str] = None,
+    **kwargs: t.Any,
+) -> None:
     """Compile SQL and execute against the current target database."""
-    _get_dbt_operations(ctx, vars).run(**kwargs)
+    _get_dbt_operations(ctx, vars).run(environment=env, **kwargs)
 
 
 @dbt.command(name="list")

From 27d1f3de64626ebf1c753f38fc4a1b2dfa96ceb9 Mon Sep 17 00:00:00 2001
From: Erin Drummond <erin.dru@gmail.com>
Date: Thu, 28 Aug 2025 04:44:41 +0000
Subject: [PATCH 6/7] Remove usage of no_gaps

---
 sqlmesh/core/context.py   |  7 +++++--
 sqlmesh_dbt/operations.py | 19 +++++++++++--------
 tests/cli/test_cli.py     |  3 +--
 3 files changed, 17 insertions(+), 12 deletions(-)

diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py
index 1a5375183c..4c1ffb1e92 100644
--- a/sqlmesh/core/context.py
+++ b/sqlmesh/core/context.py
@@ -1518,8 +1518,11 @@ def plan_builder(
             include_unmodified = self.config.plan.include_unmodified
 
         if skip_backfill and not no_gaps and not is_dev:
-            raise ConfigError(
-                "When targeting the production environment either the backfill should not be skipped or the lack of data gaps should be enforced (--no-gaps flag)."
+            # note: we deliberately don't mention the --no-gaps flag in case the plan came from the sqlmesh_dbt command
+            # todo: perhaps we could have better error messages if we check sys.argv[0] for which cli is running?
+            self.console.log_warning(
+                "Skipping the backfill stage for production can lead to unexpected results, such as tables being empty or incremental data with non-contiguous time ranges being made available.\n"
+                "If you are doing this deliberately to create an empty version of a table to test a change, please consider using Virtual Data Environments instead."
             )
 
         if not skip_linter:
diff --git a/sqlmesh_dbt/operations.py b/sqlmesh_dbt/operations.py
index beec6af068..ac7ad031f3 100644
--- a/sqlmesh_dbt/operations.py
+++ b/sqlmesh_dbt/operations.py
@@ -127,15 +127,12 @@ def _plan_options(
             )
 
         if empty:
-            # dbt --empty adds LIMIT 0 to the queries, resulting in empty tables
-            # this lines up with --skip-backfill in SQLMesh
+            # `dbt --empty` adds LIMIT 0 to the queries, resulting in empty tables. In addition, it happily clobbers existing tables regardless of if they are populated.
+            # This *partially* lines up with --skip-backfill in SQLMesh, which indicates to not populate tables if they happened to be created/updated as part of this plan.
+            # However, if a table already exists and has data in it, there is no change so SQLMesh will not recreate the table and thus it will not be cleared.
+            # So in order to fully replicate dbt's --empty, we also need --full-refresh semantics in order to replace existing tables
             options["skip_backfill"] = True
-
-            if is_prod:
-                # to prevent the following error:
-                # > ConfigError: When targeting the production environment either the backfill should not be skipped or
-                # > the lack of data gaps should be enforced (--no-gaps flag).
-                options["no_gaps"] = True
+            full_refresh = True
 
         if full_refresh:
             # TODO: handling this requires some updates in the engine to enable restatements+changes in the same plan without affecting prod
@@ -186,6 +183,12 @@ def create(
         from sqlmesh_dbt.console import DbtCliConsole
         from sqlmesh.utils.errors import SQLMeshError
 
+        # clear any existing handlers set up by click/rich as defaults so that once SQLMesh logging config is applied,
+        # we dont get duplicate messages logged from things like console.log_warning()
+        root_logger = logging.getLogger()
+        while root_logger.hasHandlers():
+            root_logger.removeHandler(root_logger.handlers[0])
+
         configure_logging(force_debug=debug)
         set_console(DbtCliConsole())
 
diff --git a/tests/cli/test_cli.py b/tests/cli/test_cli.py
index 433e2165d8..581c31af80 100644
--- a/tests/cli/test_cli.py
+++ b/tests/cli/test_cli.py
@@ -261,8 +261,7 @@ def test_plan_skip_backfill(runner, tmp_path, flag):
     result = runner.invoke(cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", flag])
     assert result.exit_code == 1
     assert (
-        "Error: When targeting the production environment either the backfill should not be skipped or the lack of data gaps should be enforced (--no-gaps flag)."
-        in result.output
+        "Skipping the backfill stage for production can lead to unexpected results" in result.output
     )
 
     # plan executes virtual update without executing model batches

From ace10d7edf3374fa14b1a7df945ece05c85c61d7 Mon Sep 17 00:00:00 2001
From: Erin Drummond <erin.dru@gmail.com>
Date: Thu, 28 Aug 2025 05:06:31 +0000
Subject: [PATCH 7/7] Adjust test for CI console text wrapping

---
 tests/cli/test_cli.py | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/tests/cli/test_cli.py b/tests/cli/test_cli.py
index 581c31af80..ef5b80e151 100644
--- a/tests/cli/test_cli.py
+++ b/tests/cli/test_cli.py
@@ -260,9 +260,7 @@ def test_plan_skip_backfill(runner, tmp_path, flag):
     # plan for `prod` errors if `--skip-backfill` is passed without --no-gaps
     result = runner.invoke(cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", flag])
     assert result.exit_code == 1
-    assert (
-        "Skipping the backfill stage for production can lead to unexpected results" in result.output
-    )
+    assert "Skipping the backfill stage for production can lead to unexpected" in result.output
 
     # plan executes virtual update without executing model batches
     # Input: `y` to perform virtual update