diff --git a/src/sqlbuild/cli/commands/_helpers/build_planning/defer_clone.py b/src/sqlbuild/cli/commands/_helpers/build_planning/defer_clone.py index 9ecf6a3c2c..f2cac8ecb2 100644 --- a/src/sqlbuild/cli/commands/_helpers/build_planning/defer_clone.py +++ b/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, ) ) diff --git a/src/sqlbuild/cli/commands/_helpers/build_planning/retention_decrease.py b/src/sqlbuild/cli/commands/_helpers/build_planning/retention_decrease.py new file mode 100644 index 0000000000..2ae5bda812 --- /dev/null +++ b/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)) diff --git a/src/sqlbuild/cli/commands/_helpers/build_virtual/virtual_execution.py b/src/sqlbuild/cli/commands/_helpers/build_virtual/virtual_execution.py index fa9d95378f..965b60e741 100644 --- a/src/sqlbuild/cli/commands/_helpers/build_virtual/virtual_execution.py +++ b/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, diff --git a/src/sqlbuild/cli/commands/_helpers/clone/execution.py b/src/sqlbuild/cli/commands/_helpers/clone/execution.py index 56e20f2e63..dc7899c07c 100644 --- a/src/sqlbuild/cli/commands/_helpers/clone/execution.py +++ b/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, ) ) diff --git a/src/sqlbuild/cli/commands/_helpers/dbt/sqlbuild_work.py b/src/sqlbuild/cli/commands/_helpers/dbt/sqlbuild_work.py index 4205d0eb49..f38709f558 100644 --- a/src/sqlbuild/cli/commands/_helpers/dbt/sqlbuild_work.py +++ b/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 ) diff --git a/src/sqlbuild/cli/commands/_helpers/entry/dispatch.py b/src/sqlbuild/cli/commands/_helpers/entry/dispatch.py index 51afc3d890..d9022f4410 100644 --- a/src/sqlbuild/cli/commands/_helpers/entry/dispatch.py +++ b/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, diff --git a/src/sqlbuild/cli/commands/_helpers/entry/parser_arguments.py b/src/sqlbuild/cli/commands/_helpers/entry/parser_arguments.py index 594adb5f34..73036ec8e2 100644 --- a/src/sqlbuild/cli/commands/_helpers/entry/parser_arguments.py +++ b/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) diff --git a/src/sqlbuild/cli/commands/classes/cli_namespace.py b/src/sqlbuild/cli/commands/classes/cli_namespace.py index 40a9b59ca3..3b81968715 100644 --- a/src/sqlbuild/cli/commands/classes/cli_namespace.py +++ b/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 diff --git a/src/sqlbuild/cli/commands/classes/virtual_build_plan_hook.py b/src/sqlbuild/cli/commands/classes/virtual_build_plan_hook.py index 557f857af6..dc003570f2 100644 --- a/src/sqlbuild/cli/commands/classes/virtual_build_plan_hook.py +++ b/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, diff --git a/src/sqlbuild/cli/commands/main/execution/_build.py b/src/sqlbuild/cli/commands/main/execution/_build.py index 9b552e796e..baa3137623 100644 --- a/src/sqlbuild/cli/commands/main/execution/_build.py +++ b/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, diff --git a/src/sqlbuild/cli/commands/models.py b/src/sqlbuild/cli/commands/models.py index 311b6682ed..4b1893ca87 100644 --- a/src/sqlbuild/cli/commands/models.py +++ b/src/sqlbuild/cli/commands/models.py @@ -173,6 +173,7 @@ class BuildCommandRequest: include_python: bool = True allow_snapshot_full_refresh: bool = False allow_table_type_downgrade: bool = False + allow_retention_decrease: bool = False allow_snapshot_schema_change: bool = False concurrency: int | None = None select: tuple[str, ...] = () @@ -302,6 +303,7 @@ class VirtualBuildPlanHookConfig: full_refresh: bool allow_snapshot_full_refresh: bool allow_table_type_downgrade: bool + allow_retention_decrease: bool use_color: bool verbose: bool debug: bool @@ -337,6 +339,7 @@ class VirtualBuildCliRequest: fail_fast: bool = False allow_snapshot_full_refresh: bool = False allow_table_type_downgrade: bool = False + allow_retention_decrease: bool = False allow_snapshot_schema_change: bool = False concurrency: int | None = None verbose: bool = False diff --git a/src/sqlbuild/cli/output/_helpers/plan_json.py b/src/sqlbuild/cli/output/_helpers/plan_json.py index 87bdcb74ee..23381f91b7 100644 --- a/src/sqlbuild/cli/output/_helpers/plan_json.py +++ b/src/sqlbuild/cli/output/_helpers/plan_json.py @@ -58,6 +58,7 @@ def format_plan_json( "phase": entry.phase.value, "statements": entry.statements, "irreversible_warning": entry.irreversible_warning, + "decrease_policy": entry.decrease_policy, } for entry in plan.retention_entries ] diff --git a/src/sqlbuild/compiler/compile/_helpers/config/retention.py b/src/sqlbuild/compiler/compile/_helpers/config/retention.py index 8ab610ebaf..8c14e0b4be 100644 --- a/src/sqlbuild/compiler/compile/_helpers/config/retention.py +++ b/src/sqlbuild/compiler/compile/_helpers/config/retention.py @@ -45,6 +45,14 @@ def resolve_time_travel_retention( if materialization_policy is not None: policy = materialization_policy source = TimeTravelRetentionSource.MATERIALIZATION + target_materialization_policy: AuthoredTimeTravelRetention | None = ( + target_config.time_travel_retention_by_materialization.get(materialized) + if target_config is not None + else None + ) + if target_materialization_policy is not None: + policy = target_materialization_policy + source = TimeTravelRetentionSource.TARGET if model_value is not None: if not isinstance(model_value, str): raise CompileInputError( diff --git a/src/sqlbuild/compiler/discovery/_helpers/yml/project.py b/src/sqlbuild/compiler/discovery/_helpers/yml/project.py index 8f724df335..01a4e9b1c1 100644 --- a/src/sqlbuild/compiler/discovery/_helpers/yml/project.py +++ b/src/sqlbuild/compiler/discovery/_helpers/yml/project.py @@ -79,6 +79,7 @@ EventExportSeverity, FutureCursorAction, MicrobatchLimitAction, + RetentionDecreasePolicy, TableType, TableTypeDowngradePolicy, TableTypeValue, @@ -987,6 +988,63 @@ def _optional_table_type_downgrade_policy( ) from exc +def _optional_target_retention_default( + *, mapping: dict[str, object], target_name: str, file_path: Path +) -> AuthoredTimeTravelRetention | None: + if isinstance(mapping.get("time_travel_retention"), dict): + return None + return _optional_retention_policy( + mapping=mapping, + key="time_travel_retention", + label=f"targets.{target_name}.time_travel_retention", + file_path=file_path, + ) + + +def _target_retention_by_materialization( + *, mapping: dict[str, object], target_name: str, file_path: Path +) -> dict[str, AuthoredTimeTravelRetention]: + payload: object | None = mapping.get("time_travel_retention") + if not isinstance(payload, dict): + return {} + label: str = f"targets.{target_name}.time_travel_retention" + retention_mapping: dict[str, object] = _coerce_mapping( + payload=payload, label=label, file_path=file_path + ) + _validate_allowed_keys( + mapping=retention_mapping, + allowed_keys=frozenset(TIME_TRAVEL_RETENTION_MATERIALIZATIONS), + label=label, + file_path=file_path, + ) + loaded: dict[str, AuthoredTimeTravelRetention] = {} + for materialization in TIME_TRAVEL_RETENTION_MATERIALIZATIONS: + policy: AuthoredTimeTravelRetention | None = _optional_retention_policy( + mapping=retention_mapping, + key=materialization, + label=f"{label}.{materialization}", + file_path=file_path, + ) + if policy is not None: + loaded[materialization] = policy + return loaded + + +def _optional_retention_decrease_policy( + *, mapping: dict[str, object], target_name: str, file_path: Path +) -> RetentionDecreasePolicy | None: + value: object | None = mapping.get("time_travel_retention_decrease") + if value is None: + return None + try: + return RetentionDecreasePolicy(value) + except (TypeError, ValueError) as exc: + raise ProjectConfigError( + f"{file_path} targets.{target_name}.time_travel_retention_decrease must be " + "'deny', 'require_confirmation', or 'allow'" + ) from exc + + def _load_targets(*, payload: object, file_path: Path) -> dict[str, TargetConfig]: mapping: dict[str, object] = _coerce_mapping( payload=payload, label="targets", file_path=file_path @@ -1038,11 +1096,11 @@ def _load_targets(*, payload: object, file_path: Path) -> dict[str, TargetConfig defer_clone_from=_optional_str(payload=target_mapping, key="defer_clone_from"), changes_only=_optional_nullable_bool(mapping=target_mapping, key="changes_only"), compile_cache=_optional_nullable_bool(mapping=target_mapping, key="compile_cache"), - time_travel_retention=_optional_retention_policy( - mapping=target_mapping, - key="time_travel_retention", - label=f"targets.{target_name}.time_travel_retention", - file_path=file_path, + time_travel_retention=_optional_target_retention_default( + mapping=target_mapping, target_name=target_name, file_path=file_path + ), + time_travel_retention_by_materialization=_target_retention_by_materialization( + mapping=target_mapping, target_name=target_name, file_path=file_path ), owns_time_travel_retention_namespace=_optional_bool( mapping=target_mapping, @@ -1064,6 +1122,12 @@ def _load_targets(*, payload: object, file_path: Path) -> dict[str, TargetConfig ) or TableTypeDowngradePolicy.REQUIRE_CONFIRMATION ), + time_travel_retention_decrease=( + _optional_retention_decrease_policy( + mapping=target_mapping, target_name=target_name, file_path=file_path + ) + or RetentionDecreasePolicy.DENY + ), execution_limits=execution_limits, clone=ClonePolicy( allow_as_clone_origin=_optional_bool( @@ -1146,11 +1210,15 @@ def _load_local_targets(*, payload: object, file_path: Path) -> dict[str, LocalT defer_clone_from=_optional_str(payload=target_mapping, key="defer_clone_from"), changes_only=_optional_nullable_bool(mapping=target_mapping, key="changes_only"), compile_cache=_optional_nullable_bool(mapping=target_mapping, key="compile_cache"), - time_travel_retention=_optional_retention_policy( - mapping=target_mapping, - key="time_travel_retention", - label=f"targets.{target_name}.time_travel_retention", - file_path=file_path, + time_travel_retention=_optional_target_retention_default( + mapping=target_mapping, target_name=target_name, file_path=file_path + ), + time_travel_retention_by_materialization=( + _target_retention_by_materialization( + mapping=target_mapping, target_name=target_name, file_path=file_path + ) + if isinstance(target_mapping.get("time_travel_retention"), dict) + else None ), owns_time_travel_retention_namespace=_optional_nullable_bool( mapping=target_mapping, @@ -1168,6 +1236,9 @@ def _load_local_targets(*, payload: object, file_path: Path) -> dict[str, LocalT label=f"targets.{target_name}.table_type_downgrade", file_path=file_path, ), + time_travel_retention_decrease=_optional_retention_decrease_policy( + mapping=target_mapping, target_name=target_name, file_path=file_path + ), execution_limits=execution_limits, clone=LocalClonePolicy( allow_as_clone_origin=_optional_nullable_bool( @@ -1245,6 +1316,7 @@ def _validate_target_keys( "owns_time_travel_retention_namespace", "default_table_type", "table_type_downgrade", + "time_travel_retention_decrease", "clone", "state", "execution_limits", diff --git a/src/sqlbuild/compiler/planner/_helpers/planning/retention.py b/src/sqlbuild/compiler/planner/_helpers/planning/retention.py index fa28169ac3..0b6e5ee402 100644 --- a/src/sqlbuild/compiler/planner/_helpers/planning/retention.py +++ b/src/sqlbuild/compiler/planner/_helpers/planning/retention.py @@ -4,6 +4,7 @@ import hashlib from collections import defaultdict +from dataclasses import replace from sqlbuild.adapter.contract.models import ( RelationInfo, @@ -130,12 +131,16 @@ def plan_retention( ) if not selected_models: return () - if runtime.adapter.adapter_name == BuiltinAdapter.BIGQUERY: - return _plan_bigquery_retention(runtime=runtime, models=selected_models) entries: list[RetentionPlanEntry] = [] - for model in selected_models: - entries.extend(_plan_relation_retention(runtime=runtime, warehouse=warehouse, model=model)) - return tuple(entries) + if runtime.adapter.adapter_name == BuiltinAdapter.BIGQUERY: + entries.extend(_plan_bigquery_retention(runtime=runtime, models=selected_models)) + else: + for model in selected_models: + entries.extend( + _plan_relation_retention(runtime=runtime, warehouse=warehouse, model=model) + ) + decrease_policy: str = _effective_target(runtime=runtime).time_travel_retention_decrease.value + return tuple(replace(entry, decrease_policy=decrease_policy) for entry in entries) def _plan_relation_retention( diff --git a/src/sqlbuild/compiler/planner/models.py b/src/sqlbuild/compiler/planner/models.py index 121bfdb6d6..3b4ad820f5 100644 --- a/src/sqlbuild/compiler/planner/models.py +++ b/src/sqlbuild/compiler/planner/models.py @@ -954,6 +954,13 @@ class RetentionPlanEntry: phase: RetentionPlanPhase statements: tuple[str, ...] = field(default_factory=tuple) irreversible_warning: str | None = None + decrease_policy: str = "deny" + + @property + def decreases(self) -> bool: + """Whether this entry lowers any live retention value.""" + + return self.direction in (RetentionDirection.DECREASE, RetentionDirection.MIXED) @dataclass(frozen=True) diff --git a/src/sqlbuild/executor/build/_helpers/retention.py b/src/sqlbuild/executor/build/_helpers/retention.py index 81bbaa22a0..b93639fd0b 100644 --- a/src/sqlbuild/executor/build/_helpers/retention.py +++ b/src/sqlbuild/executor/build/_helpers/retention.py @@ -16,7 +16,12 @@ RetentionPlanEntry, TableTypePlanEntry, ) -from sqlbuild.compiler.planner.types import IncrementalMode, PlanAction, RetentionPlanPhase +from sqlbuild.compiler.planner.types import ( + IncrementalMode, + PlanAction, + RetentionDirection, + RetentionPlanPhase, +) from sqlbuild.errors.contracts.exceptions import ExecutorInputError from sqlbuild.runtime.observability.classes.operation_lifecycle import ( OperationAttributes, @@ -202,6 +207,11 @@ def reconcile_retention_after_build( ) -> None: """Converge all remaining retention drift after the full build succeeds.""" + recreated_model_names: frozenset[str] = frozenset( + model_entry.name + for model_entry in plan.model_entries + if materialization_recreates_relation(model_entry) + ) for entry in plan.retention_entries: if entry.phase == RetentionPlanPhase.NONE: continue @@ -214,7 +224,12 @@ def reconcile_retention_after_build( request=entry.request, state=state, ) + lowering_permitted: bool = _lowering_permitted( + entry=entry, state=state, recreated_model_names=recreated_model_names + ) for change in changes: + if not lowering_permitted and change.phase != RetentionChangePhase.PREPARE: + continue _apply_retention_statements( adapter=adapter, connection=connection, @@ -223,6 +238,27 @@ def reconcile_retention_after_build( ) +def _lowering_permitted( + *, entry: RetentionPlanEntry, state: RetentionState, recreated_model_names: frozenset[str] +) -> bool: + """Only lower live retention for gated planned decreases or relations created this build.""" + + if entry.decreases or entry.direction == RetentionDirection.APPLY_AFTER_CREATE: + return True + if entry.model_names and all(name in recreated_model_names for name in entry.model_names): + return True + desired_days: int = entry.request.desired_days + values: tuple[int, ...] = tuple( + value + for value in ( + state.delta_log_retention_days, + state.delta_deleted_file_retention_days, + ) + if value is not None + ) or (state.effective_days,) + return all(value <= desired_days for value in values) + + def _safe_before_build_success( *, entry: RetentionPlanEntry, state: RetentionState, change: RenderedRetentionChange ) -> bool: diff --git a/src/sqlbuild/executor/clone/_helpers/lifecycle.py b/src/sqlbuild/executor/clone/_helpers/lifecycle.py index 5e5a5e319e..8cb75027d1 100644 --- a/src/sqlbuild/executor/clone/_helpers/lifecycle.py +++ b/src/sqlbuild/executor/clone/_helpers/lifecycle.py @@ -49,9 +49,13 @@ def prepare_clone_destination( def finish_clone( *, results: list[CloneItemResult], inputs: CloneExecutionInput ) -> CloneExecutionResult: - """Apply namespace decreases only after every clone item succeeds.""" + """Apply permitted namespace decreases only after every clone item succeeds.""" - if results and all(result.status == CloneStatus.SUCCESS for result in results): + if ( + inputs.allow_namespace_retention_decrease + and results + and all(result.status == CloneStatus.SUCCESS for result in results) + ): _ = apply_clone_namespace_retention_phase( requests=inputs.destination_retention_requests, adapter=inputs.adapter, diff --git a/src/sqlbuild/executor/clone/models.py b/src/sqlbuild/executor/clone/models.py index f98fa4eee0..524ddda90b 100644 --- a/src/sqlbuild/executor/clone/models.py +++ b/src/sqlbuild/executor/clone/models.py @@ -48,6 +48,7 @@ class CloneExecutionInput: default_factory=dict ) destination_retention_requests: dict[str, RetentionRequest] = field(default_factory=dict) + allow_namespace_retention_decrease: bool = False on_item: CloneItemCallback | None = None diff --git a/src/sqlbuild/spec/contracts/_helpers/targets.py b/src/sqlbuild/spec/contracts/_helpers/targets.py index 28d8f58468..1113741dfd 100644 --- a/src/sqlbuild/spec/contracts/_helpers/targets.py +++ b/src/sqlbuild/spec/contracts/_helpers/targets.py @@ -49,6 +49,10 @@ def resolve_target_config( local_target: LocalTargetConfig | None = local_config.targets.get(target_name) if local_target is None: return project_target + local_overrides_retention: bool = ( + local_target.time_travel_retention is not None + or local_target.time_travel_retention_by_materialization is not None + ) target_config: TargetConfig = TargetConfig( connection={**project_target.connection, **local_target.connection}, connection_name=( @@ -88,9 +92,14 @@ def resolve_target_config( ), time_travel_retention=( local_target.time_travel_retention - if local_target.time_travel_retention is not None + if local_overrides_retention else project_target.time_travel_retention ), + time_travel_retention_by_materialization=( + local_target.time_travel_retention_by_materialization or {} + if local_overrides_retention + else project_target.time_travel_retention_by_materialization + ), owns_time_travel_retention_namespace=( local_target.owns_time_travel_retention_namespace if local_target.owns_time_travel_retention_namespace is not None @@ -106,6 +115,11 @@ def resolve_target_config( if local_target.table_type_downgrade is not None else project_target.table_type_downgrade ), + time_travel_retention_decrease=( + local_target.time_travel_retention_decrease + if local_target.time_travel_retention_decrease is not None + else project_target.time_travel_retention_decrease + ), execution_limits=_merge_execution_limits( project_limits=project_target.execution_limits, local_limits=local_target.execution_limits, diff --git a/src/sqlbuild/spec/contracts/models.py b/src/sqlbuild/spec/contracts/models.py index 855b357027..9052288303 100644 --- a/src/sqlbuild/spec/contracts/models.py +++ b/src/sqlbuild/spec/contracts/models.py @@ -14,6 +14,7 @@ EventExportSeverity, FutureCursorAction, MicrobatchLimitAction, + RetentionDecreasePolicy, SourceFreshnessStrategy, SourceFreshnessValueKind, SourceWriteStrategy, @@ -139,9 +140,13 @@ class TargetConfig: state: StateConfig = field(default_factory=StateConfig) compile_cache: bool | None = None time_travel_retention: AuthoredTimeTravelRetention | None = None + time_travel_retention_by_materialization: dict[str, AuthoredTimeTravelRetention] = field( + default_factory=dict + ) owns_time_travel_retention_namespace: bool = False default_table_type: TableType | None = None table_type_downgrade: TableTypeDowngradePolicy = TableTypeDowngradePolicy.REQUIRE_CONFIRMATION + time_travel_retention_decrease: RetentionDecreasePolicy = RetentionDecreasePolicy.DENY execution_limits: ExecutionLimitsConfig = field(default_factory=ExecutionLimitsConfig) @@ -162,9 +167,11 @@ class LocalTargetConfig: state: LocalStateConfig = field(default_factory=LocalStateConfig) compile_cache: bool | None = None time_travel_retention: AuthoredTimeTravelRetention | None = None + time_travel_retention_by_materialization: dict[str, AuthoredTimeTravelRetention] | None = None owns_time_travel_retention_namespace: bool | None = None default_table_type: TableType | None = None table_type_downgrade: TableTypeDowngradePolicy | None = None + time_travel_retention_decrease: RetentionDecreasePolicy | None = None execution_limits: ExecutionLimitsConfig = field(default_factory=ExecutionLimitsConfig) diff --git a/src/sqlbuild/spec/contracts/types.py b/src/sqlbuild/spec/contracts/types.py index f1c53763b4..9c174147ec 100644 --- a/src/sqlbuild/spec/contracts/types.py +++ b/src/sqlbuild/spec/contracts/types.py @@ -101,3 +101,11 @@ class TableTypeDowngradePolicy(StrEnum): DENY = "deny" REQUIRE_CONFIRMATION = "require_confirmation" ALLOW = "allow" + + +class RetentionDecreasePolicy(StrEnum): + """Policy for lowering live time-travel retention.""" + + DENY = "deny" + REQUIRE_CONFIRMATION = "require_confirmation" + ALLOW = "allow" diff --git a/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py b/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py index d8066b1362..36f29da476 100644 --- a/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py +++ b/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py @@ -577,3 +577,19 @@ class TargetRetentionViewsTestCase: view_header: str expected_exit_code: int expected_fragment: str + + +@dataclass(frozen=True) +class RetentionDecreasePolicyTestCase: + description: str + target_lines: tuple[str, ...] + build_flags: tuple[str, ...] + expected_exit_code: int + expected_fragment: str + + +@dataclass(frozen=True) +class TargetMaterializationRetentionTestCase: + description: str + target_lines: tuple[str, ...] + expected_requested_days: frozenset[int] diff --git a/tests/integration/src/sqlbuild/cli/commands/main/helpers.py b/tests/integration/src/sqlbuild/cli/commands/main/helpers.py index 9f080e40e5..f8f8786c90 100644 --- a/tests/integration/src/sqlbuild/cli/commands/main/helpers.py +++ b/tests/integration/src/sqlbuild/cli/commands/main/helpers.py @@ -8,6 +8,13 @@ import duckdb import pytest +from sqlbuild.adapter.contract.models import ( + RenderedRetentionChange, + RetentionRequest, + RetentionState, +) +from sqlbuild.adapter.contract.types import RetentionChangePhase +from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter from sqlbuild.cli.commands.main.entrypoint.entry import main @@ -151,3 +158,91 @@ def compile_finding_keys(*, project_dir: Path, capsys: pytest.CaptureFixture[str payload: dict[str, object] = json.loads(capsys.readouterr().out) diagnostics: list[dict[str, object]] = cast(list[dict[str, object]], payload["diagnostics"]) return {f"{item['path']}:{item['code']}" for item in diagnostics} + + +class LiveRetentionFake: + """Pretend every managed table currently keeps a fixed number of retention days.""" + + def __init__(self, *, live_days: int) -> None: + self.live_days: int = live_days + self.requested_days: list[int] = [] + + def inspect_retention(self, *, connection: object, request: RetentionRequest) -> RetentionState: + del connection + return RetentionState( + request_id=request.request_id, + scope=request.scope, + configured_days=self.live_days, + effective_days=self.live_days, + ) + + def render_retention_changes( + self, *, request: RetentionRequest, state: RetentionState | None = None + ) -> tuple[RenderedRetentionChange, ...]: + del state + self.requested_days.append(request.desired_days) + return ( + RenderedRetentionChange(phase=RetentionChangePhase.ALTER, statements=("SELECT 1",)), + ) + + +def install_live_retention_fake( + *, monkeypatch: pytest.MonkeyPatch, fake: LiveRetentionFake +) -> None: + """Route DuckDB retention inspection and rendering through the fake.""" + + monkeypatch.setattr( + DuckDbAdapter, + "inspect_retention", + lambda _self, *, connection, request: fake.inspect_retention( + connection=connection, request=request + ), + ) + monkeypatch.setattr( + DuckDbAdapter, + "render_retention_changes", + lambda _self, *, request, state=None: fake.render_retention_changes( + request=request, state=state + ), + ) + + +def write_retention_policy_project(*, project_dir: Path, target_lines: tuple[str, ...]) -> None: + """Write a DuckDB project with one table and one view under a prod target.""" + + (project_dir / "models").mkdir(parents=True, exist_ok=True) + _ = (project_dir / "sqlbuild_project.toml").write_text( + "\n".join( + ( + 'name = "demo"', + 'adapter = "duckdb"', + 'default_target = "prod"', + "", + "[connection]", + 'database = "demo.duckdb"', + "", + "[materialization_defaults.table]", + 'time_travel_retention = "90d"', + "", + "[targets.prod]", + *target_lines, + "", + ) + ), + encoding="utf-8", + ) + _ = (project_dir / "models" / "orders.sql").write_text( + "MODEL (materialized table);\n\nSELECT 1 AS order_id\n", encoding="utf-8" + ) + _ = (project_dir / "models" / "order_view.sql").write_text( + 'MODEL (materialized view);\n\nSELECT order_id FROM __ref("orders")\n', encoding="utf-8" + ) + + +def run_build( + *, project_dir: Path, flags: tuple[str, ...], capsys: pytest.CaptureFixture[str] +) -> tuple[int, str]: + """Run one real CLI build and return its exit code and combined output.""" + + exit_code: int = main(["--project-dir", str(project_dir), "--no-color", "build", *flags]) + return exit_code, "".join(capsys.readouterr()) diff --git a/tests/integration/src/sqlbuild/cli/commands/main/test_retention_policies.py b/tests/integration/src/sqlbuild/cli/commands/main/test_retention_policies.py new file mode 100644 index 0000000000..dcf0ec0c8f --- /dev/null +++ b/tests/integration/src/sqlbuild/cli/commands/main/test_retention_policies.py @@ -0,0 +1,124 @@ +"""Integration coverage for target retention policies through real CLI builds.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from tests.integration.src.sqlbuild.cli.commands.main._test_types import ( + RetentionDecreasePolicyTestCase, + TargetMaterializationRetentionTestCase, +) +from tests.integration.src.sqlbuild.cli.commands.main.helpers import ( + LiveRetentionFake, + install_live_retention_fake, + run_build, + write_retention_policy_project, +) + + +@pytest.mark.parametrize( + "test_case", + [ + RetentionDecreasePolicyTestCase( + description="decrease is denied by default", + target_lines=('time_travel_retention = { table = "7d" }',), + build_flags=("--allow-retention-decrease",), + expected_exit_code=1, + expected_fragment="time travel retention decrease is denied for 'orders'", + ), + RetentionDecreasePolicyTestCase( + description="confirmation policy fails non-interactive builds without the flag", + target_lines=( + 'time_travel_retention = { table = "7d" }', + 'time_travel_retention_decrease = "require_confirmation"', + ), + build_flags=(), + expected_exit_code=1, + expected_fragment="--allow-retention-decrease", + ), + RetentionDecreasePolicyTestCase( + description="confirmation policy proceeds with the flag", + target_lines=( + 'time_travel_retention = { table = "7d" }', + 'time_travel_retention_decrease = "require_confirmation"', + ), + build_flags=("--allow-retention-decrease",), + expected_exit_code=0, + expected_fragment="Completed successfully", + ), + RetentionDecreasePolicyTestCase( + description="allow policy proceeds silently", + target_lines=( + 'time_travel_retention = { table = "7d" }', + 'time_travel_retention_decrease = "allow"', + ), + build_flags=(), + expected_exit_code=0, + expected_fragment="Completed successfully", + ), + RetentionDecreasePolicyTestCase( + description="increase is never gated", + target_lines=('time_travel_retention = { table = "120d" }',), + build_flags=(), + expected_exit_code=0, + expected_fragment="Completed successfully", + ), + ], + ids=lambda case: case.description, +) +def test_given_live_retention_when_rebuilding_with_new_target_retention_then_policy_is_enforced( + test_case: RetentionDecreasePolicyTestCase, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + install_live_retention_fake(monkeypatch=monkeypatch, fake=LiveRetentionFake(live_days=90)) + write_retention_policy_project( + project_dir=tmp_path, target_lines=('time_travel_retention = "90d"',) + ) + first_exit, first_output = run_build(project_dir=tmp_path, flags=(), capsys=capsys) + assert first_exit == 0, first_output + + write_retention_policy_project(project_dir=tmp_path, target_lines=test_case.target_lines) + exit_code, output = run_build(project_dir=tmp_path, flags=test_case.build_flags, capsys=capsys) + + assert exit_code == test_case.expected_exit_code, output + assert test_case.expected_fragment in output + + +@pytest.mark.parametrize( + "test_case", + [ + TargetMaterializationRetentionTestCase( + description="target materialization retention beats project materialization default", + target_lines=('time_travel_retention = { table = "120d" }',), + expected_requested_days=frozenset({120}), + ), + TargetMaterializationRetentionTestCase( + description="project materialization default beats target-wide default", + target_lines=('time_travel_retention = "30d"',), + expected_requested_days=frozenset({90}), + ), + ], + ids=lambda case: case.description, +) +def test_given_target_materialization_retention_when_building_then_precedence_is_applied( + test_case: TargetMaterializationRetentionTestCase, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + fake: LiveRetentionFake = LiveRetentionFake(live_days=1) + install_live_retention_fake(monkeypatch=monkeypatch, fake=fake) + write_retention_policy_project(project_dir=tmp_path, target_lines=test_case.target_lines) + + exit_code, output = run_build(project_dir=tmp_path, flags=(), capsys=capsys) + + assert exit_code == 0, output + assert frozenset(fake.requested_days) == test_case.expected_requested_days + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"]) diff --git a/tests/unit/src/sqlbuild/cli/commands/_helpers/build_planning/_test_types.py b/tests/unit/src/sqlbuild/cli/commands/_helpers/build_planning/_test_types.py index 59e15dc1d8..9c030abe85 100644 --- a/tests/unit/src/sqlbuild/cli/commands/_helpers/build_planning/_test_types.py +++ b/tests/unit/src/sqlbuild/cli/commands/_helpers/build_planning/_test_types.py @@ -45,3 +45,15 @@ class UnsupportedDurationLimitTestCase: remediation: str expected_code: str expected_error_fragment: str + + +@dataclass(frozen=True) +class RetentionDecreasePolicyTestCase: + description: str + plan_output: PlanOutput + allow_retention_decrease: bool + expected_error_fragment: str | None = None + expected_help_fragment: str = "" + expected_output: str = "" + input_text: str = "" + input_is_tty: bool = False diff --git a/tests/unit/src/sqlbuild/cli/commands/_helpers/build_planning/helpers.py b/tests/unit/src/sqlbuild/cli/commands/_helpers/build_planning/helpers.py index 6ee48230d7..a56e607ba3 100644 --- a/tests/unit/src/sqlbuild/cli/commands/_helpers/build_planning/helpers.py +++ b/tests/unit/src/sqlbuild/cli/commands/_helpers/build_planning/helpers.py @@ -2,10 +2,22 @@ from pathlib import Path +from sqlbuild.adapter.contract.models import RetentionRequest +from sqlbuild.adapter.contract.types import RetentionScope from sqlbuild.compiler.compile.models import CompiledObjectKey, CompiledRelationLocation from sqlbuild.compiler.compile.types import CompiledResourceType -from sqlbuild.compiler.planner.models import ModelPlanEntry, TableTypePlanEntry -from sqlbuild.compiler.planner.types import MaterializationType, PlanAction, PlanReason +from sqlbuild.compiler.planner.models import ( + ModelPlanEntry, + RetentionPlanEntry, + TableTypePlanEntry, +) +from sqlbuild.compiler.planner.types import ( + MaterializationType, + PlanAction, + PlanReason, + RetentionDirection, + RetentionPlanPhase, +) def build_snapshot_full_refresh_entry( @@ -58,3 +70,29 @@ def build_table_type_entry( downgrade=downgrade, downgrade_policy=policy, ) + + +def build_retention_entry( + *, + name: str = "orders", + direction: RetentionDirection = RetentionDirection.DECREASE, + policy: str = "deny", +) -> RetentionPlanEntry: + return RetentionPlanEntry( + request=RetentionRequest( + request_id=name, + scope=RetentionScope.RELATION, + database=None, + schema="main", + name=name, + desired_days=7, + ), + model_names=(name,), + actual_days=90, + effective_days=90, + source="target", + direction=direction, + phase=RetentionPlanPhase.POST, + statements=("SELECT 1",), + decrease_policy=policy, + ) diff --git a/tests/unit/src/sqlbuild/cli/commands/_helpers/build_planning/test_retention_decrease.py b/tests/unit/src/sqlbuild/cli/commands/_helpers/build_planning/test_retention_decrease.py new file mode 100644 index 0000000000..563da4b4fc --- /dev/null +++ b/tests/unit/src/sqlbuild/cli/commands/_helpers/build_planning/test_retention_decrease.py @@ -0,0 +1,144 @@ +from __future__ import annotations + +from io import StringIO + +import pytest + +from sqlbuild.cli.commands._helpers.build_planning.retention_decrease import ( + enforce_retention_decrease_policy, +) +from sqlbuild.cli.commands.exceptions import CliUserError +from sqlbuild.compiler.planner.models import PlanOutput +from sqlbuild.compiler.planner.types import RetentionDirection +from tests.unit.src.sqlbuild.cli.commands._helpers.build_planning._test_types import ( + RetentionDecreasePolicyTestCase, +) +from tests.unit.src.sqlbuild.cli.commands._helpers.build_planning.helpers import ( + build_retention_entry, +) + + +class _InputStream(StringIO): + def __init__(self, initial_value: str, *, is_tty: bool) -> None: + super().__init__(initial_value) + self._is_tty = is_tty + + def isatty(self) -> bool: + return self._is_tty + + +@pytest.mark.parametrize( + "test_case", + [ + RetentionDecreasePolicyTestCase( + description="deny lists affected models even with the CLI flag", + plan_output=PlanOutput( + retention_entries=( + build_retention_entry(name="orders"), + build_retention_entry(name="customers"), + ) + ), + allow_retention_decrease=True, + expected_error_fragment="'orders', 'customers'", + expected_help_fragment="time_travel_retention_decrease", + ), + RetentionDecreasePolicyTestCase( + description="mixed direction counts as a decrease", + plan_output=PlanOutput( + retention_entries=(build_retention_entry(direction=RetentionDirection.MIXED),) + ), + allow_retention_decrease=False, + expected_error_fragment="denied for 'orders'", + expected_help_fragment="time_travel_retention_decrease", + ), + RetentionDecreasePolicyTestCase( + description="non-interactive confirmation requires CLI flag", + plan_output=PlanOutput( + retention_entries=(build_retention_entry(policy="require_confirmation"),) + ), + allow_retention_decrease=False, + expected_error_fragment="requires confirmation", + expected_help_fragment="--allow-retention-decrease", + ), + RetentionDecreasePolicyTestCase( + description="incorrect interactive phrase cancels", + plan_output=PlanOutput( + retention_entries=(build_retention_entry(policy="require_confirmation"),) + ), + allow_retention_decrease=False, + expected_error_fragment="cancelled", + input_text="decrease everything\n", + input_is_tty=True, + ), + ], + ids=lambda case: case.description, +) +def test_given_unsafe_retention_decrease_when_enforcing_then_raises_user_error( + test_case: RetentionDecreasePolicyTestCase, +) -> None: + with pytest.raises(CliUserError) as exc_info: + enforce_retention_decrease_policy( + plan=test_case.plan_output, + allow_retention_decrease=test_case.allow_retention_decrease, + input_stream=_InputStream(test_case.input_text, is_tty=test_case.input_is_tty), + output_stream=StringIO(), + ) + + error: CliUserError = exc_info.value + assert test_case.expected_error_fragment is not None + assert test_case.expected_error_fragment in error.message + assert test_case.expected_help_fragment in (error.help or "") + + +@pytest.mark.parametrize( + "test_case", + [ + RetentionDecreasePolicyTestCase( + description="CLI flag confirms required decrease", + plan_output=PlanOutput( + retention_entries=(build_retention_entry(policy="require_confirmation"),) + ), + allow_retention_decrease=True, + ), + RetentionDecreasePolicyTestCase( + description="allow policy proceeds silently", + plan_output=PlanOutput(retention_entries=(build_retention_entry(policy="allow"),)), + allow_retention_decrease=False, + ), + RetentionDecreasePolicyTestCase( + description="increase is never gated", + plan_output=PlanOutput( + retention_entries=(build_retention_entry(direction=RetentionDirection.INCREASE),) + ), + allow_retention_decrease=False, + ), + RetentionDecreasePolicyTestCase( + description="exact interactive phrase confirms decrease", + plan_output=PlanOutput( + retention_entries=(build_retention_entry(policy="require_confirmation"),) + ), + allow_retention_decrease=False, + expected_output=( + "Decreasing time travel retention for 'orders' permanently discards history " + "older than the new retention.\n\n" + "Type `decrease retention for orders` to continue: " + ), + input_text="decrease retention for orders\n", + input_is_tty=True, + ), + ], + ids=lambda case: case.description, +) +def test_given_safe_or_confirmed_retention_change_when_enforcing_then_allows_execution( + test_case: RetentionDecreasePolicyTestCase, +) -> None: + output_stream: StringIO = StringIO() + + enforce_retention_decrease_policy( + plan=test_case.plan_output, + allow_retention_decrease=test_case.allow_retention_decrease, + input_stream=_InputStream(test_case.input_text, is_tty=test_case.input_is_tty), + output_stream=output_stream, + ) + + assert output_stream.getvalue() == test_case.expected_output diff --git a/tests/unit/src/sqlbuild/compiler/discovery/_helpers/_test_types.py b/tests/unit/src/sqlbuild/compiler/discovery/_helpers/_test_types.py index 1203be7ccb..7463f1b6b3 100644 --- a/tests/unit/src/sqlbuild/compiler/discovery/_helpers/_test_types.py +++ b/tests/unit/src/sqlbuild/compiler/discovery/_helpers/_test_types.py @@ -2,6 +2,8 @@ from decimal import Decimal from pathlib import Path +from sqlbuild.spec.contracts.models import AuthoredTimeTravelRetention +from sqlbuild.spec.contracts.types import RetentionDecreasePolicy from sqlbuild.sql_values.types import CollectionRendering @@ -801,3 +803,19 @@ class ExpectedCountTestCase: class ExpectedMessageTestCase: description: str expected_message: str + + +@dataclass(frozen=True) +class LoadTargetRetentionPoliciesTestCase: + description: str + target_lines: tuple[str, ...] + expected_default: AuthoredTimeTravelRetention | None + expected_by_materialization: dict[str, AuthoredTimeTravelRetention] + expected_decrease_policy: RetentionDecreasePolicy + + +@dataclass(frozen=True) +class LoadTargetRetentionPoliciesErrorTestCase: + description: str + target_lines: tuple[str, ...] + expected_error_fragment: str diff --git a/tests/unit/src/sqlbuild/compiler/discovery/_helpers/test_yml_project.py b/tests/unit/src/sqlbuild/compiler/discovery/_helpers/test_yml_project.py index abb2f76a3e..11adb31dd0 100644 --- a/tests/unit/src/sqlbuild/compiler/discovery/_helpers/test_yml_project.py +++ b/tests/unit/src/sqlbuild/compiler/discovery/_helpers/test_yml_project.py @@ -11,7 +11,14 @@ ) from sqlbuild.compiler.discovery.exceptions import ProjectConfigError from sqlbuild.runtime.event_exporting.types import LifecycleEventKind -from sqlbuild.spec.contracts.models import ExecutionLimitsConfig, LocalConfig, ProjectConfig +from sqlbuild.spec.contracts.models import ( + AuthoredTimeTravelRetention, + ExecutionLimitsConfig, + LocalConfig, + ProjectConfig, + TargetConfig, +) +from sqlbuild.spec.contracts.types import RetentionDecreasePolicy from sqlbuild.sql_values.types import CollectionRendering from tests.unit.src.sqlbuild.compiler.discovery._helpers._test_types import ( ColumnContractModeConfigErrorTestCase, @@ -31,6 +38,8 @@ LoadProjectCostConfigTestCase, LoadRetentionConfigErrorTestCase, LoadRetentionConfigTestCase, + LoadTargetRetentionPoliciesErrorTestCase, + LoadTargetRetentionPoliciesTestCase, MicrobatchLimitConfigErrorTestCase, MicrobatchLimitConfigTestCase, StartCursorConfigTestCase, @@ -506,6 +515,86 @@ def test_given_retention_config_when_loading_project_then_policies_are_typed( ) +@pytest.mark.parametrize( + "test_case", + [ + LoadTargetRetentionPoliciesTestCase( + description="string retention is a target-wide default and decreases are denied", + target_lines=('time_travel_retention = "30d"',), + expected_default=AuthoredTimeTravelRetention(desired_days=30), + expected_by_materialization={}, + expected_decrease_policy=RetentionDecreasePolicy.DENY, + ), + LoadTargetRetentionPoliciesTestCase( + description="retention table sets per-materialization values", + target_lines=( + 'time_travel_retention = { table = "90d", snapshot = "365d" }', + 'time_travel_retention_decrease = "require_confirmation"', + ), + expected_default=None, + expected_by_materialization={ + "table": AuthoredTimeTravelRetention(desired_days=90), + "snapshot": AuthoredTimeTravelRetention(desired_days=365), + }, + expected_decrease_policy=RetentionDecreasePolicy.REQUIRE_CONFIRMATION, + ), + ], + ids=lambda case: case.description, +) +def test_given_target_retention_policies_when_loading_project_then_they_are_typed( + test_case: LoadTargetRetentionPoliciesTestCase, + tmp_path: Path, +) -> None: + (tmp_path / "sqlbuild_project.toml").write_text( + "\n".join( + ('name = "demo"', 'adapter = "snowflake"', "[targets.prod]", *test_case.target_lines) + ), + encoding="utf-8", + ) + + target: TargetConfig = load_project_config(project_dir=tmp_path).targets["prod"] + + assert target.time_travel_retention == test_case.expected_default + assert target.time_travel_retention_by_materialization == test_case.expected_by_materialization + assert target.time_travel_retention_decrease is test_case.expected_decrease_policy + + +@pytest.mark.parametrize( + "test_case", + [ + LoadTargetRetentionPoliciesErrorTestCase( + description="retention table rejects non-table materializations", + target_lines=('time_travel_retention = { view = "7d" }',), + expected_error_fragment="view", + ), + LoadTargetRetentionPoliciesErrorTestCase( + description="retention table values must be whole days", + target_lines=('time_travel_retention = { table = "12h" }',), + expected_error_fragment="targets.prod.time_travel_retention.table", + ), + LoadTargetRetentionPoliciesErrorTestCase( + description="decrease policy must be a known value", + target_lines=('time_travel_retention_decrease = "sometimes"',), + expected_error_fragment="'deny', 'require_confirmation', or 'allow'", + ), + ], + ids=lambda case: case.description, +) +def test_given_invalid_target_retention_policies_when_loading_project_then_config_error_is_raised( + test_case: LoadTargetRetentionPoliciesErrorTestCase, + tmp_path: Path, +) -> None: + (tmp_path / "sqlbuild_project.toml").write_text( + "\n".join( + ('name = "demo"', 'adapter = "snowflake"', "[targets.prod]", *test_case.target_lines) + ), + encoding="utf-8", + ) + + with pytest.raises(ProjectConfigError, match=test_case.expected_error_fragment): + load_project_config(project_dir=tmp_path) + + @pytest.mark.parametrize( "test_case", [ diff --git a/tests/unit/src/sqlbuild/executor/build/_helpers/_test_types.py b/tests/unit/src/sqlbuild/executor/build/_helpers/_test_types.py index 1ba7f16451..eda6e0c24b 100644 --- a/tests/unit/src/sqlbuild/executor/build/_helpers/_test_types.py +++ b/tests/unit/src/sqlbuild/executor/build/_helpers/_test_types.py @@ -4,7 +4,12 @@ from sqlbuild.adapter.contract.models import RelationInfo from sqlbuild.adapter.contract.types import RetentionChangePhase from sqlbuild.compiler.discovery.models import DiscoveredLoaderFunction -from sqlbuild.compiler.planner.types import MaterializationType, PlanAction, RetentionPlanPhase +from sqlbuild.compiler.planner.types import ( + MaterializationType, + PlanAction, + RetentionDirection, + RetentionPlanPhase, +) from sqlbuild.executor.build.models import BuildExecutionResult from sqlbuild.executor.build.types import BuildStatus from sqlbuild.executor.scheduling.types import ExecutionStatus @@ -70,6 +75,16 @@ class BuildModelRetentionReconciliationTestCase: expected_statements: tuple[str, ...] +@dataclass(frozen=True) +class FinalRetentionReconciliationTestCase: + description: str + planned_direction: RetentionDirection + desired_days: int + live_days: int + expected_statements: tuple[str, ...] + model_action: PlanAction = PlanAction.INCREMENTAL_MERGE + + @dataclass(frozen=True) class TableTypeConversionTestCase: description: str diff --git a/tests/unit/src/sqlbuild/executor/build/_helpers/test_retention.py b/tests/unit/src/sqlbuild/executor/build/_helpers/test_retention.py index 8dc8e77c72..0fa70c0382 100644 --- a/tests/unit/src/sqlbuild/executor/build/_helpers/test_retention.py +++ b/tests/unit/src/sqlbuild/executor/build/_helpers/test_retention.py @@ -32,11 +32,13 @@ apply_table_type_conversion, materialization_recreates_relation, reconcile_model_retention, + reconcile_retention_after_build, ) from sqlbuild.observability import EventDispatcher, LifecycleEvent, dispatcher_scope from tests.unit.src.sqlbuild.executor.build._helpers._test_types import ( BuildModelRetentionReconciliationTestCase, BuildRetentionPhaseTestCase, + FinalRetentionReconciliationTestCase, LifecycleProgressTestCase, MaterializationRecreatesRelationTestCase, TableTypeConversionErrorTestCase, @@ -211,6 +213,92 @@ def test_given_successful_model_when_reconciling_retention_then_defers_decreases ] +@pytest.mark.parametrize( + "test_case", + [ + FinalRetentionReconciliationTestCase( + description="gated planned decrease is applied", + planned_direction=RetentionDirection.DECREASE, + desired_days=7, + live_days=30, + expected_statements=("ALTER RETENTION",), + ), + FinalRetentionReconciliationTestCase( + description="new relation is lowered to its desired retention", + planned_direction=RetentionDirection.APPLY_AFTER_CREATE, + desired_days=7, + live_days=30, + expected_statements=("ALTER RETENTION",), + ), + FinalRetentionReconciliationTestCase( + description="retention raised after planning an increase is never lowered", + planned_direction=RetentionDirection.INCREASE, + desired_days=7, + live_days=30, + expected_statements=(), + ), + FinalRetentionReconciliationTestCase( + description="recreated table is lowered from its inherited default", + planned_direction=RetentionDirection.INCREASE, + desired_days=7, + live_days=30, + expected_statements=("ALTER RETENTION",), + model_action=PlanAction.CREATE_TABLE, + ), + FinalRetentionReconciliationTestCase( + description="planned increase still converges upward", + planned_direction=RetentionDirection.INCREASE, + desired_days=7, + live_days=1, + expected_statements=("ALTER RETENTION",), + ), + ], + ids=lambda case: case.description, +) +def test_given_live_retention_after_build_when_reconciling_then_only_gated_decreases_lower_it( + test_case: FinalRetentionReconciliationTestCase, +) -> None: + request: RetentionRequest = RetentionRequest( + request_id="orders", + scope=RetentionScope.RELATION, + database=None, + schema="analytics", + name="orders", + desired_days=test_case.desired_days, + ) + plan: PlanOutput = PlanOutput( + retention_entries=( + RetentionPlanEntry( + request=request, + model_names=("orders",), + actual_days=None, + effective_days=None, + source="target", + direction=test_case.planned_direction, + phase=RetentionPlanPhase.POST, + ), + ), + model_entries=(build_model_plan_entry(name="orders", action=test_case.model_action),), + ) + adapter: Mock = Mock() + adapter.inspect_retention.return_value = RetentionState( + request_id="orders", + scope=RetentionScope.RELATION, + configured_days=test_case.live_days, + effective_days=test_case.live_days, + ) + adapter.render_retention_changes.return_value = ( + RenderedRetentionChange(phase=RetentionChangePhase.ALTER, statements=("ALTER RETENTION",)), + ) + connection: object = object() + + reconcile_retention_after_build(plan=plan, adapter=adapter, connection=connection) + + assert adapter.execute.call_args_list == [ + call(connection=connection, sql=statement) for statement in test_case.expected_statements + ] + + @pytest.mark.parametrize( "test_case", [ diff --git a/tests/unit/src/sqlbuild/executor/clone/_helpers/_test_types.py b/tests/unit/src/sqlbuild/executor/clone/_helpers/_test_types.py index 48e20a4c3e..90323e58b9 100644 --- a/tests/unit/src/sqlbuild/executor/clone/_helpers/_test_types.py +++ b/tests/unit/src/sqlbuild/executor/clone/_helpers/_test_types.py @@ -39,3 +39,10 @@ class CloneRetentionSelectionTestCase: description: str selected_model_names: frozenset[str] expected_request_names: tuple[str, ...] + + +@dataclass(frozen=True) +class CloneNamespaceDecreasePolicyTestCase: + description: str + allow_namespace_retention_decrease: bool + expected_statements: tuple[str, ...] diff --git a/tests/unit/src/sqlbuild/executor/clone/_helpers/test_lifecycle.py b/tests/unit/src/sqlbuild/executor/clone/_helpers/test_lifecycle.py new file mode 100644 index 0000000000..ab112798a4 --- /dev/null +++ b/tests/unit/src/sqlbuild/executor/clone/_helpers/test_lifecycle.py @@ -0,0 +1,87 @@ +from __future__ import annotations + +from unittest.mock import Mock, call + +import pytest + +from sqlbuild.adapter.contract.models import ( + RenderedRetentionChange, + RetentionRequest, + RetentionState, +) +from sqlbuild.adapter.contract.types import RetentionChangePhase, RetentionScope +from sqlbuild.executor.clone._helpers.lifecycle import finish_clone +from sqlbuild.executor.clone.models import CloneExecutionInput, CloneItemResult, CloneSourceEntries +from sqlbuild.executor.clone.types import CloneAction, CloneStatus +from tests.unit.src.sqlbuild.executor.clone._helpers._test_types import ( + CloneNamespaceDecreasePolicyTestCase, +) + + +@pytest.mark.parametrize( + "test_case", + [ + CloneNamespaceDecreasePolicyTestCase( + description="namespace decrease is skipped unless the target allows it", + allow_namespace_retention_decrease=False, + expected_statements=(), + ), + CloneNamespaceDecreasePolicyTestCase( + description="allowed namespace decrease is applied after clone success", + allow_namespace_retention_decrease=True, + expected_statements=("ALTER DATASET RETENTION 2",), + ), + ], + ids=lambda case: case.description, +) +def test_given_namespace_retention_decrease_when_finishing_clone_then_policy_is_respected( + test_case: CloneNamespaceDecreasePolicyTestCase, +) -> None: + adapter: Mock = Mock() + adapter.inspect_retention.return_value = RetentionState( + request_id="analytics", + scope=RetentionScope.NAMESPACE, + configured_days=7, + effective_days=7, + ) + adapter.render_retention_changes.return_value = ( + RenderedRetentionChange( + phase=RetentionChangePhase.ALTER, statements=("ALTER DATASET RETENTION 2",) + ), + ) + connection: object = object() + inputs: CloneExecutionInput = CloneExecutionInput( + source_entries=CloneSourceEntries(), + origin_model_entries=(), + destination_model_entries=(), + origin_seed_entries=(), + destination_seed_entries=(), + destination_function_entries=(), + execution_order=(), + adapter=adapter, + destination_connection=connection, + hard_copy=False, + run_id="run_1", + query_change_tracking=False, + destination_retention_requests={ + "orders": RetentionRequest( + request_id="analytics", + scope=RetentionScope.NAMESPACE, + database="warehouse", + schema="analytics", + desired_days=2, + ) + }, + allow_namespace_retention_decrease=test_case.allow_namespace_retention_decrease, + ) + + _ = finish_clone( + results=[ + CloneItemResult(name="orders", action=CloneAction.CLONED, status=CloneStatus.SUCCESS) + ], + inputs=inputs, + ) + + assert adapter.execute.call_args_list == [ + call(connection=connection, sql=statement) for statement in test_case.expected_statements + ] diff --git a/tests/unit/src/sqlbuild/spec/contracts/main/_test_types.py b/tests/unit/src/sqlbuild/spec/contracts/main/_test_types.py index 3234635f21..621a08611a 100644 --- a/tests/unit/src/sqlbuild/spec/contracts/main/_test_types.py +++ b/tests/unit/src/sqlbuild/spec/contracts/main/_test_types.py @@ -1,8 +1,9 @@ from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, field from sqlbuild.spec.contracts.models import ( + AuthoredTimeTravelRetention, ExecutionLimitsConfig, LoaderDestinationParts, LocalConfig, @@ -50,7 +51,10 @@ class TargetRetentionResolutionTestCase: project_config: ProjectConfig local_config: LocalConfig target_name: str - expected_desired_days: int + expected_default: AuthoredTimeTravelRetention | None + expected_by_materialization: dict[str, AuthoredTimeTravelRetention] = field( + default_factory=dict + ) @dataclass(frozen=True) diff --git a/tests/unit/src/sqlbuild/spec/contracts/main/test_targets.py b/tests/unit/src/sqlbuild/spec/contracts/main/test_targets.py index e1ef3e8dce..3f6bccbbae 100644 --- a/tests/unit/src/sqlbuild/spec/contracts/main/test_targets.py +++ b/tests/unit/src/sqlbuild/spec/contracts/main/test_targets.py @@ -222,8 +222,32 @@ def test_given_project_and_local_state_config_when_resolving_then_local_override } ), target_name="dev", - expected_desired_days=2, - ) + expected_default=AuthoredTimeTravelRetention(desired_days=2), + ), + TargetRetentionResolutionTestCase( + description="local retention table replaces project target default", + project_config=ProjectConfig( + name="demo", + adapter="snowflake", + targets={ + "dev": TargetConfig( + time_travel_retention=AuthoredTimeTravelRetention(desired_days=7) + ) + }, + ), + local_config=LocalConfig( + targets={ + "dev": LocalTargetConfig( + time_travel_retention_by_materialization={ + "table": AuthoredTimeTravelRetention(desired_days=1) + } + ) + } + ), + target_name="dev", + expected_default=None, + expected_by_materialization={"table": AuthoredTimeTravelRetention(desired_days=1)}, + ), ], ids=lambda case: case.description, ) @@ -236,8 +260,11 @@ def test_given_local_target_retention_when_resolving_then_it_overrides_project_t target_name=test_case.target_name, ) - assert target_config.time_travel_retention is not None - assert target_config.time_travel_retention.desired_days == test_case.expected_desired_days + assert target_config.time_travel_retention == test_case.expected_default + assert ( + target_config.time_travel_retention_by_materialization + == test_case.expected_by_materialization + ) @pytest.mark.parametrize(