feat: deny time travel retention decreases by default and support per-materialization target retention (#795)

* feat: deny time travel retention decreases by default and support per-materialization target retention

* fix: never lower retention outside gated planned decreases

* fix: lower retention on tables recreated during the build

Author: kvlonge

src/sqlbuild/cli/commands/_helpers/build_planning/defer_clone.py

@@ -58,6 +58,7 @@
 )
 from sqlbuild.executor.clone.types import CloneStatus
 from sqlbuild.spec.contracts.main.resolve_target_config import resolve_target_config
+from sqlbuild.spec.contracts.types import RetentionDecreasePolicy
 
 
 def build_defer_clone_boundary_selectors(
@@ -304,6 +305,14 @@ def run_clone(on_item: Any) -> CloneExecutionResult:
                             target_name=destination_target_name,
                         ).owns_time_travel_retention_namespace,
                     ),
+                    allow_namespace_retention_decrease=(
+                        resolve_target_config(
+                            project_config=discovered_inputs.project_config,
+                            local_config=discovered_inputs.local_config,
+                            target_name=destination_target_name,
+                        ).time_travel_retention_decrease
+                        == RetentionDecreasePolicy.ALLOW
+                    ),
                     on_item=on_item,
                 )
             )

src/sqlbuild/cli/commands/_helpers/build_planning/retention_decrease.py

@@ -0,0 +1,73 @@
+"""Time-travel retention decrease safety enforcement for build commands."""
+
+from __future__ import annotations
+
+from typing import TextIO
+
+from sqlbuild.cli.commands.exceptions import CliUserError
+from sqlbuild.compiler.planner.models import PlanOutput, RetentionPlanEntry
+from sqlbuild.spec.contracts.types import RetentionDecreasePolicy
+
+
+def enforce_retention_decrease_policy(
+    *,
+    plan: PlanOutput,
+    allow_retention_decrease: bool,
+    input_stream: TextIO,
+    output_stream: TextIO,
+) -> None:
+    """Fail or confirm before lowering live time-travel retention."""
+
+    entries: tuple[RetentionPlanEntry, ...] = tuple(
+        entry for entry in plan.retention_entries if entry.decreases
+    )
+    denied: tuple[RetentionPlanEntry, ...] = tuple(
+        entry for entry in entries if entry.decrease_policy == RetentionDecreasePolicy.DENY
+    )
+    if denied:
+        raise CliUserError(
+            f"time travel retention decrease is denied for {_names(denied)}",
+            help=(
+                "Keep the current retention, or set the target's "
+                "time_travel_retention_decrease to 'require_confirmation' or 'allow'."
+            ),
+        )
+    confirmation: tuple[RetentionPlanEntry, ...] = tuple(
+        entry
+        for entry in entries
+        if entry.decrease_policy == RetentionDecreasePolicy.REQUIRE_CONFIRMATION
+    )
+    if not confirmation or allow_retention_decrease:
+        return
+    if not input_stream.isatty():
+        raise CliUserError(
+            "time travel retention decrease requires confirmation",
+            help="Pass --allow-retention-decrease to confirm in non-interactive runs.",
+        )
+    expected: str = _confirmation_text(confirmation)
+    output_stream.write(
+        f"Decreasing time travel retention for {_names(confirmation)} permanently discards "
+        "history older than the new retention.\n\n"
+    )
+    output_stream.write(f"Type `{expected}` to continue: ")
+    output_stream.flush()
+    if input_stream.readline().strip() != expected:
+        raise CliUserError("time travel retention decrease cancelled")
+
+
+def _model_names(entries: tuple[RetentionPlanEntry, ...]) -> tuple[str, ...]:
+    names: dict[str, None] = {}
+    for entry in entries:
+        names.update(dict.fromkeys(entry.model_names))
+    return tuple(names)
+
+
+def _confirmation_text(entries: tuple[RetentionPlanEntry, ...]) -> str:
+    names: tuple[str, ...] = _model_names(entries)
+    if len(names) == 1:
+        return f"decrease retention for {names[0]}"
+    return f"decrease retention for {len(names)} models"
+
+
+def _names(entries: tuple[RetentionPlanEntry, ...]) -> str:
+    return ", ".join(f"'{name}'" for name in _model_names(entries))

src/sqlbuild/cli/commands/_helpers/build_virtual/virtual_execution.py

@@ -96,6 +96,7 @@ def execute_virtual_build(
             full_refresh=request.full_refresh,
             allow_snapshot_full_refresh=request.allow_snapshot_full_refresh,
             allow_table_type_downgrade=request.allow_table_type_downgrade,
+            allow_retention_decrease=request.allow_retention_decrease,
             use_color=request.use_color,
             verbose=request.verbose,
             debug=request.debug,

src/sqlbuild/cli/commands/_helpers/clone/execution.py

@@ -44,6 +44,7 @@
 from sqlbuild.runtime.observability.classes.operation_lifecycle import OperationLifecycle
 from sqlbuild.spec.contracts.main.resolve_target_config import resolve_target_config
 from sqlbuild.spec.contracts.models import SourceEntry
+from sqlbuild.spec.contracts.types import RetentionDecreasePolicy
 
 
 def execute_clone_plan(
@@ -102,6 +103,14 @@ def execute_clone_plan(
                         target_name=invocation.destination_target_name,
                     ).owns_time_travel_retention_namespace,
                 ),
+                allow_namespace_retention_decrease=(
+                    resolve_target_config(
+                        project_config=invocation.discovered_inputs.project_config,
+                        local_config=invocation.discovered_inputs.local_config,
+                        target_name=invocation.destination_target_name,
+                    ).time_travel_retention_decrease
+                    == RetentionDecreasePolicy.ALLOW
+                ),
                 on_item=on_item,
             )
         )

src/sqlbuild/cli/commands/_helpers/dbt/sqlbuild_work.py

@@ -2,10 +2,14 @@
 
 from __future__ import annotations
 
+import sys
 from pathlib import Path
 from typing import TextIO
 
 from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter
+from sqlbuild.cli.commands._helpers.build_planning.retention_decrease import (
+    enforce_retention_decrease_policy,
+)
 from sqlbuild.cli.commands.classes.build_progress_callbacks import (
     BuildProgressCallbacks,
     format_build_footer,
@@ -46,6 +50,12 @@ def execute_sqlbuild_build_work(
     adapter_name: str = context.adapter_name
     output_stream: TextIO = context.output_stream
     use_color: bool = context.use_color
+    enforce_retention_decrease_policy(
+        plan=plan_output,
+        allow_retention_decrease=False,
+        input_stream=sys.stdin,
+        output_stream=output_stream,
+    )
     callbacks: BuildProgressCallbacks = BuildProgressCallbacks(
         plan=plan_output, use_color=use_color, verbose=verbose, debug=False
     )

src/sqlbuild/cli/commands/_helpers/entry/dispatch.py

@@ -165,6 +165,7 @@ def dispatch_cli_command(*, args: CliNamespace, handlers: CliEntrypointHandlers)
                 include_python=args.include_python,
                 allow_snapshot_full_refresh=args.allow_snapshot_full_refresh,
                 allow_table_type_downgrade=args.allow_table_type_downgrade,
+                allow_retention_decrease=args.allow_retention_decrease,
                 allow_snapshot_schema_change=args.allow_snapshot_schema_change,
                 concurrency=args.concurrency,
                 select=select,

src/sqlbuild/cli/commands/_helpers/entry/parser_arguments.py

@@ -62,6 +62,7 @@ def add_execution_args(parser: argparse.ArgumentParser) -> None:
     parser.add_argument("--full-refresh", action="store_true", default=False)
     parser.add_argument("--allow-snapshot-full-refresh", action="store_true", default=False)
     parser.add_argument("--allow-table-type-downgrade", action="store_true", default=False)
+    parser.add_argument("--allow-retention-decrease", action="store_true", default=False)
     parser.add_argument("--allow-snapshot-schema-change", action="store_true", default=False)
     parser.add_argument("--concurrency", type=int, default=None)
     parser.add_argument("--verbose", "-v", action="store_true", default=False)

src/sqlbuild/cli/commands/classes/cli_namespace.py

@@ -61,6 +61,7 @@
     "run_audits": True,
     "allow_snapshot_full_refresh": False,
     "allow_table_type_downgrade": False,
+    "allow_retention_decrease": False,
     "allow_snapshot_schema_change": False,
     "allow_partial_promotion": False,
     "allow_partial_rollback": False,
@@ -234,6 +235,7 @@ class CliNamespace:
     run_audits: bool
     allow_snapshot_full_refresh: bool
     allow_table_type_downgrade: bool
+    allow_retention_decrease: bool
     allow_snapshot_schema_change: bool
     allow_partial_promotion: bool
     allow_partial_rollback: bool

src/sqlbuild/cli/commands/classes/virtual_build_plan_hook.py

@@ -17,6 +17,9 @@
 from sqlbuild.cli.commands._helpers.build_planning.full_refresh import (
     enforce_snapshot_full_refresh_policy,
 )
+from sqlbuild.cli.commands._helpers.build_planning.retention_decrease import (
+    enforce_retention_decrease_policy,
+)
 from sqlbuild.cli.commands._helpers.build_planning.table_type import (
     enforce_table_type_downgrade_policy,
 )
@@ -52,6 +55,7 @@ def __init__(
         self._full_refresh = config.full_refresh
         self._allow_snapshot_full_refresh = config.allow_snapshot_full_refresh
         self._allow_table_type_downgrade = config.allow_table_type_downgrade
+        self._allow_retention_decrease = config.allow_retention_decrease
         self._use_color = config.use_color
         self._verbose = config.verbose
         self._debug = config.debug
@@ -109,6 +113,12 @@ def on_plan_ready(
             input_stream=sys.stdin,
             output_stream=sys.stdout,
         )
+        enforce_retention_decrease_policy(
+            plan=plan_output,
+            allow_retention_decrease=self._allow_retention_decrease,
+            input_stream=sys.stdin,
+            output_stream=sys.stdout,
+        )
         write_compile_target(
             target_dir=self._project_dir / "target",
             adapter=self._adapter,

src/sqlbuild/cli/commands/main/execution/_build.py

@@ -34,6 +34,9 @@
 )
 from sqlbuild.cli.commands._helpers.build_planning.invocation import resolve_build_invocation
 from sqlbuild.cli.commands._helpers.build_planning.planning import compile_build_plan
+from sqlbuild.cli.commands._helpers.build_planning.retention_decrease import (
+    enforce_retention_decrease_policy,
+)
 from sqlbuild.cli.commands._helpers.build_planning.table_type import (
     enforce_table_type_downgrade_policy,
 )
@@ -129,6 +132,7 @@ def _run_build(
                     fail_fast=request.fail_fast,
                     allow_snapshot_full_refresh=request.allow_snapshot_full_refresh,
                     allow_table_type_downgrade=request.allow_table_type_downgrade,
+                    allow_retention_decrease=request.allow_retention_decrease,
                     allow_snapshot_schema_change=request.allow_snapshot_schema_change,
                     concurrency=request.concurrency,
                     verbose=request.verbose,
@@ -177,6 +181,12 @@ def _run_build(
             input_stream=sys.stdin,
             output_stream=sys.stdout,
         )
+        enforce_retention_decrease_policy(
+            plan=pipeline_result.plan_output,
+            allow_retention_decrease=request.allow_retention_decrease,
+            input_stream=sys.stdin,
+            output_stream=sys.stdout,
+        )
         write_build_compile_target(
             request=request,
             invocation=invocation,