add more granular bot controls for preview plans

Signed-off-by: Jesse Hodges <hodges.jesse@gmail.com>

Author: gjesse

docs/integrations/github.md

@@ -293,8 +293,10 @@ Below is an example of how to define the default config for the bot in either YA
 | `enable_deploy_command`               | Indicates if the `/deploy` command should be enabled in order to allowed synchronized deploys to production. Default: `False`                                                                                                                                                                                                                                                                             |  bool  |    N     |
 | `command_namespace`                   | The namespace to use for SQLMesh commands. For example if you provide `#SQLMesh` as a value then commands will be expected in the format of `#SQLMesh/<command>`. Default: `None` meaning no namespace is used.                                                                                                                                                                                           | string |    N     |
 | `auto_categorize_changes`             | Auto categorization behavior to use for the bot. If not provided then the project-wide categorization behavior is used. See [Auto-categorize model changes](https://sqlmesh.readthedocs.io/en/stable/guides/configuration/#auto-categorize-model-changes) for details.                                                                                                                                    |  dict  |    N     |
-| `default_pr_start`                    | Default start when creating PR environment plans. If running in a mode where the bot automatically backfills models (based on `auto_categorize_changes` behavior) then this can be used to limit the amount of data backfilled. Defaults to `None` meaning the start date is set to the earliest model's start or to 1 day ago if [data previews](../concepts/plans.md#data-preview) need to be computed. |  str   |    N     |
+| `default_pr_start`                    | Default start when creating PR environment plans. If running in a mode where the bot automatically backfills models (based on `auto_categorize_changes` behavior) then this can be used to limit the amount of data backfilled. Defaults to `None` meaning the start date is set to the earliest model's start.                                                                                              |  str   |    N     |
 | `pr_min_intervals`                    | Intended for use when `default_pr_start` is set to a relative time, eg `1 week ago`. This ensures that at least this many intervals across every model are included for backfill in the PR environment. Without this, models with an interval unit wider than `default_pr_start` (such as `@monthly` models if `default_pr_start` was set to `1 week ago`) will be excluded from backfill entirely.       |  int   |    N     |
+| `default_pr_preview_start`            | Default start when computing [data previews](../concepts/plans.md#data-preview) for forward-only changes in PR environments. Defaults to `yesterday`, independent of `default_pr_start`, so preview data can be limited without reducing the regular PR backfill window.                                                                                                                                       |  str   |    N     |
+| `pr_preview_min_intervals`            | Intended for use when `default_pr_preview_start` is set to a relative time. This ensures that at least this many intervals are included for forward-only previews in the PR environment. Default: `1`                                                                                                                                                                                                          |  int   |    N     |
 | `skip_pr_backfill`                    | Indicates if the bot should skip backfilling models in the PR environment. Default: `True`                                                                                                                                                                                                                                                                                                                |  bool  |    N     |
 | `pr_include_unmodified`               | Indicates whether to include unmodified models in the PR environment. Default to the project's config value (which defaults to `False`)                                                                                                                                                                                                                                                                   |  bool  |    N     |
 | `run_on_deploy_to_prod`               | Indicates whether to run latest intervals when deploying to prod. If set to false, the deployment will backfill only the changed models up to the existing latest interval in production, ignoring any missing intervals beyond this point. Default: `False`                                                                                                                                              |  bool  |    N     |
@@ -320,6 +322,7 @@ Example with all properties defined:
         sql: full
         seed: full
       default_pr_start: "1 week ago"
+      default_pr_preview_start: "yesterday"
       skip_pr_backfill: false
       run_on_deploy_to_prod: false
       prod_branch_name: production
@@ -344,6 +347,7 @@ Example with all properties defined:
                 seed=AutoCategorizationMode.FULL,
             ),
             default_pr_start="1 week ago",
+            default_pr_preview_start="yesterday",
             skip_pr_backfill=False,
             run_on_deploy_to_prod=False,
             prod_branch_name="production",

sqlmesh/core/context.py

@@ -1469,6 +1469,8 @@ def plan_builder(
         backfill_models: t.Optional[t.Collection[str]] = None,
         categorizer_config: t.Optional[CategorizerConfig] = None,
         enable_preview: t.Optional[bool] = None,
+        preview_start: t.Optional[TimeLike] = None,
+        preview_min_intervals: t.Optional[int] = None,
         run: t.Optional[bool] = None,
         diff_rendered: t.Optional[bool] = None,
         skip_linter: t.Optional[bool] = None,
@@ -1510,6 +1512,8 @@ def plan_builder(
             select_models: A list of model selection strings to filter the models that should be included into this plan.
             backfill_models: A list of model selection strings to filter the models for which the data should be backfilled.
             enable_preview: Indicates whether to enable preview for forward-only models in development environments.
+            preview_start: The start date for forward-only previews.
+            preview_min_intervals: The minimum number of intervals to preview for each forward-only preview snapshot.
             run: Whether to run latest intervals as part of the plan application.
             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
@@ -1543,6 +1547,8 @@ def plan_builder(
             "select_models": list(select_models) if select_models is not None else None,
             "backfill_models": list(backfill_models) if backfill_models is not None else None,
             "enable_preview": enable_preview,
+            "preview_start": preview_start,
+            "preview_min_intervals": preview_min_intervals,
             "run": run,
             "diff_rendered": diff_rendered,
             "skip_linter": skip_linter,
@@ -1744,6 +1750,8 @@ def plan_builder(
             enable_preview=(
                 enable_preview if enable_preview is not None else self._plan_preview_enabled
             ),
+            preview_start=preview_start,
+            preview_min_intervals=preview_min_intervals or 0,
             end_bounded=not run,
             ensure_finalized_snapshots=self.config.plan.use_finalized_state,
             start_override_per_model=start_override_per_model,

sqlmesh/core/plan/builder.py

@@ -87,6 +87,8 @@ class PlanBuilder:
         default_start: The default plan start to use if not specified.
         default_end: The default plan end to use if not specified.
         enable_preview: Whether to enable preview for forward-only models in development environments.
+        preview_start: The start time to use for forward-only previews. Defaults to the plan start.
+        preview_min_intervals: The minimum number of intervals to preview for each forward-only preview snapshot.
         end_bounded: If set to true, the missing intervals will be bounded by the target end date, disregarding lookback,
             allow_partials, and other attributes that could cause the intervals to exceed the target end date.
         ensure_finalized_snapshots: Whether to compare against snapshots from the latest finalized
@@ -125,6 +127,8 @@ def __init__(
         default_start: t.Optional[TimeLike] = None,
         default_end: t.Optional[TimeLike] = None,
         enable_preview: bool = False,
+        preview_start: t.Optional[TimeLike] = None,
+        preview_min_intervals: int = 0,
         end_bounded: bool = False,
         ensure_finalized_snapshots: bool = False,
         explain: bool = False,
@@ -148,6 +152,8 @@ def __init__(
             allow_additive_models if allow_additive_models is not None else []
         )
         self._enable_preview = enable_preview
+        self._preview_start = preview_start
+        self._preview_min_intervals = preview_min_intervals
         self._end_bounded = end_bounded
         self._ensure_finalized_snapshots = ensure_finalized_snapshots
         self._ignore_cron = ignore_cron
@@ -447,9 +453,12 @@ def _build_restatements(
             possible_intervals = {
                 restatements[p.snapshot_id] for p in restating_parents if p.is_incremental
             }
+            removal_start = (
+                self._forward_only_preview_start(snapshot, start, end) if is_preview else start
+            )
             possible_intervals.add(
                 snapshot.get_removal_interval(
-                    start,
+                    removal_start,
                     end,
                     self._execution_time,
                     strict=False,
@@ -474,6 +483,21 @@ def _build_restatements(
 
         return restatements
 
+    def _forward_only_preview_start(
+        self, snapshot: Snapshot, default_start: TimeLike, end: TimeLike
+    ) -> TimeLike:
+        preview_start = self._preview_start or default_start
+        if not self._preview_min_intervals:
+            return preview_start
+
+        relative_base = to_datetime(self.execution_time)
+        preview_end = to_datetime(end, relative_base=relative_base)
+        min_start = snapshot.node.cron_floor(preview_end)
+        for _ in range(self._preview_min_intervals):
+            min_start = snapshot.node.cron_prev(min_start)
+
+        return min(to_datetime(preview_start, relative_base=relative_base), min_start)
+
     def _build_directly_and_indirectly_modified(
         self, dag: DAG[SnapshotId]
     ) -> t.Tuple[t.Set[SnapshotId], SnapshotMapping]:

sqlmesh/integrations/github/cicd/config.py

@@ -27,11 +27,13 @@ class GithubCICDBotConfig(BaseConfig):
         default=None, alias="auto_categorize_changes"
     )
     default_pr_start: t.Optional[TimeLike] = None
+    default_pr_preview_start: TimeLike = "yesterday"
     skip_pr_backfill_: t.Optional[bool] = Field(default=None, alias="skip_pr_backfill")
     pr_include_unmodified_: t.Optional[bool] = Field(default=None, alias="pr_include_unmodified")
     run_on_deploy_to_prod: bool = False
     pr_environment_name: t.Optional[str] = None
     pr_min_intervals: t.Optional[int] = None
+    pr_preview_min_intervals: int = Field(default=1, ge=0)
     prod_branch_names_: t.Optional[str] = Field(default=None, alias="prod_branch_name")
     forward_only_branch_suffix_: t.Optional[str] = Field(
         default=None, alias="forward_only_branch_suffix"
@@ -88,9 +90,11 @@ def forward_only_branch_suffix(self) -> str:
         "command_namespace",
         "auto_categorize_changes",
         "default_pr_start",
+        "default_pr_preview_start",
         "skip_pr_backfill",
         "pr_include_unmodified",
         "run_on_deploy_to_prod",
         "pr_min_intervals",
+        "pr_preview_min_intervals",
         "forward_only_branch_suffix",
     }

sqlmesh/integrations/github/cicd/controller.py

@@ -404,6 +404,8 @@ def pr_plan(self) -> Plan:
                 categorizer_config=self.bot_config.auto_categorize_changes,
                 start=self.bot_config.default_pr_start,
                 min_intervals=self.bot_config.pr_min_intervals,
+                preview_start=self.bot_config.default_pr_preview_start,
+                preview_min_intervals=self.bot_config.pr_preview_min_intervals,
                 skip_backfill=self.bot_config.skip_pr_backfill,
                 include_unmodified=self.bot_config.pr_include_unmodified,
                 forward_only=self.forward_only_plan,

tests/core/test_plan.py

@@ -2934,6 +2934,122 @@ def test_unaligned_start_model_with_forward_only_preview(make_snapshot):
     assert not plan.deployability_index.is_deployable(snapshot_b)
 
 
+def _make_forward_only_preview_context_diff(make_snapshot):
+    old_snapshot = make_snapshot(
+        SqlModel(
+            name="a",
+            query=parse_one("select 1, ds"),
+            kind=dict(
+                name=ModelKindName.INCREMENTAL_BY_TIME_RANGE,
+                forward_only=True,
+                time_column="ds",
+            ),
+            start="2025-01-01",
+        )
+    )
+    old_snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+
+    new_snapshot = make_snapshot(
+        SqlModel(
+            name="a",
+            query=parse_one("select 2, ds"),
+            kind=dict(
+                name=ModelKindName.INCREMENTAL_BY_TIME_RANGE,
+                forward_only=True,
+                time_column="ds",
+            ),
+            start="2025-01-01",
+        )
+    )
+    new_snapshot.previous_versions = old_snapshot.all_versions
+
+    context_diff = ContextDiff(
+        environment="test_environment",
+        is_new_environment=True,
+        is_unfinalized_environment=False,
+        normalize_environment_name=True,
+        create_from="prod",
+        create_from_env_exists=True,
+        added=set(),
+        removed_snapshots={},
+        snapshots={new_snapshot.snapshot_id: new_snapshot},
+        new_snapshots={new_snapshot.snapshot_id: new_snapshot},
+        modified_snapshots={old_snapshot.name: (new_snapshot, old_snapshot)},
+        previous_plan_id=None,
+        previously_promoted_snapshot_ids=set(),
+        previous_finalized_snapshots=None,
+        previous_gateway_managed_virtual_layer=False,
+        gateway_managed_virtual_layer=False,
+        environment_statements=[],
+    )
+
+    return context_diff, new_snapshot
+
+
+@time_machine.travel("2026-06-02 00:00:00 UTC")
+def test_forward_only_preview_uses_preview_start(make_snapshot):
+    context_diff, new_snapshot = _make_forward_only_preview_context_diff(make_snapshot)
+
+    plan = PlanBuilder(
+        context_diff,
+        start="2025-01-01",
+        preview_start="yesterday",
+        preview_min_intervals=1,
+        enable_preview=True,
+        is_dev=True,
+    ).build()
+
+    assert plan.start == "2025-01-01"
+    assert plan.restatements == {
+        new_snapshot.snapshot_id: (
+            to_timestamp("2026-06-01"),
+            to_timestamp("2026-06-02"),
+        )
+    }
+
+
+@time_machine.travel("2026-06-02 00:00:00 UTC")
+def test_forward_only_preview_min_intervals_expands_preview_start(make_snapshot):
+    context_diff, new_snapshot = _make_forward_only_preview_context_diff(make_snapshot)
+
+    plan = PlanBuilder(
+        context_diff,
+        start="2025-01-01",
+        preview_start="now",
+        preview_min_intervals=1,
+        enable_preview=True,
+        is_dev=True,
+    ).build()
+
+    assert plan.restatements == {
+        new_snapshot.snapshot_id: (
+            to_timestamp("2026-06-01"),
+            to_timestamp("2026-06-02"),
+        )
+    }
+
+
+@time_machine.travel("2026-06-02 00:00:00 UTC")
+def test_forward_only_preview_start_can_exceed_preview_min_intervals(make_snapshot):
+    context_diff, new_snapshot = _make_forward_only_preview_context_diff(make_snapshot)
+
+    plan = PlanBuilder(
+        context_diff,
+        start="2025-01-01",
+        preview_start="2025-01-01",
+        preview_min_intervals=1,
+        enable_preview=True,
+        is_dev=True,
+    ).build()
+
+    assert plan.restatements == {
+        new_snapshot.snapshot_id: (
+            to_timestamp("2025-01-01"),
+            to_timestamp("2026-06-02"),
+        )
+    }
+
+
 def test_restate_production_model_in_dev(make_snapshot, mocker: MockerFixture):
     snapshot = make_snapshot(
         SqlModel(