diff --git a/sqlmesh/core/plan/stages.py b/sqlmesh/core/plan/stages.py index 729e1705b4..0972187faa 100644 --- a/sqlmesh/core/plan/stages.py +++ b/sqlmesh/core/plan/stages.py @@ -313,7 +313,7 @@ def build(self, plan: EvaluatablePlan) -> t.List[PlanStage]: if plan.new_snapshots: stages.append(CreateSnapshotRecordsStage(snapshots=plan.new_snapshots)) - snapshots_to_create = self._get_snapshots_to_create(plan, snapshots) + snapshots_to_create = self._get_snapshots_to_create(plan, snapshots, deployability_index) if snapshots_to_create: stages.append( PhysicalLayerSchemaCreationStage( @@ -629,7 +629,7 @@ def _get_audit_only_snapshots( return audit_snapshots def _get_snapshots_to_create( - self, plan: EvaluatablePlan, snapshots: t.Dict[SnapshotId, Snapshot] + self, plan: EvaluatablePlan, snapshots: t.Dict[SnapshotId, Snapshot], deployability_index: DeployabilityIndex ) -> t.List[Snapshot]: promoted_snapshot_ids = ( set(plan.environment.promoted_snapshot_ids) @@ -640,6 +640,11 @@ def _get_snapshots_to_create( def _should_create(s: Snapshot) -> bool: if not s.is_model or s.is_symbolic: return False + + # Do not create snapshots that contain production data but can never be deployed. + if deployability_index.is_representative(s) and not deployability_index.is_deployable(s): + return False + # Only create tables for snapshots that we're planning to promote or that were selected for backfill return ( plan.is_selected_for_backfill(s.name) diff --git a/sqlmesh/core/snapshot/definition.py b/sqlmesh/core/snapshot/definition.py index 0c9635a7c2..0d2cc1eed4 100644 --- a/sqlmesh/core/snapshot/definition.py +++ b/sqlmesh/core/snapshot/definition.py @@ -1047,6 +1047,13 @@ def missing_intervals( return [] deployability_index = deployability_index or DeployabilityIndex.all_deployable() + # If a snapshot is representative but not deployable (e.g. INDIRECT_NON_BREAKING snapshots) then it will only + # ever refer to the prod physical table, and so should never have any missing intervals. + if deployability_index.is_representative(self) and not deployability_index.is_deployable( + self + ): + return [] + intervals = ( self.intervals if deployability_index.is_representative(self) else self.dev_intervals ) @@ -1610,7 +1617,7 @@ def is_deployable(self, snapshot: SnapshotIdLike) -> bool: def is_representative(self, snapshot: SnapshotIdLike) -> bool: """Returns true if the deployable (non-dev) table of the given snapshot should be used for reading, table mapping, and - computing missing intervals. + computing missing intervals (if the snapshot is also deployable). Note, that deployable snapshots are also representative, but the reverse is not always true. @@ -2063,6 +2070,13 @@ def missing_intervals( if not snapshot.evaluatable: continue + # If a snapshot is representative but not deployable (e.g. INDIRECT_NON_BREAKING snapshots) then it will only + # ever refer to the prod physical table, and so should never have any missing intervals. + if deployability_index.is_representative( + snapshot + ) and not deployability_index.is_deployable(snapshot): + continue + snapshot_start_date = start_override_per_model.get(snapshot.name, start_dt) snapshot_end_date: TimeLike = end_date diff --git a/tests/core/test_snapshot.py b/tests/core/test_snapshot.py index 64bee7f472..02936a6bad 100644 --- a/tests/core/test_snapshot.py +++ b/tests/core/test_snapshot.py @@ -638,6 +638,71 @@ def test_missing_intervals_start_override_per_model(make_snapshot: t.Callable[.. ] +def test__missing_intervals__representative_snapshot_prod_intervals_returned(snapshot: Snapshot): + # Arrange + snapshot.add_interval(start="2020-01-01", end="2020-01-01", is_dev=False) + snapshot.add_interval(start="2020-01-02", end="2020-01-02", is_dev=False) + snapshot.add_interval(start="2020-01-01", end="2020-01-01", is_dev=True) + + snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + deployability_index = DeployabilityIndex.create([snapshot]) + + # Act + missing_intervals = snapshot.missing_intervals( + start="2020-01-01", end="2020-01-03", deployability_index=deployability_index + ) + + # Assert + assert deployability_index.is_representative(snapshot) + assert missing_intervals == [ + (to_timestamp("2020-01-03"), to_timestamp("2020-01-04")), + ] + + +def test__missing_intervals__non_representative_snapshot_dev_intervals_returned(snapshot: Snapshot): + # Arrange + snapshot.add_interval(start="2020-01-01", end="2020-01-01", is_dev=False) + snapshot.add_interval(start="2020-01-02", end="2020-01-02", is_dev=False) + snapshot.add_interval(start="2020-01-01", end="2020-01-01", is_dev=True) + + snapshot.categorize_as(SnapshotChangeCategory.BREAKING, forward_only=True) + deployability_index = DeployabilityIndex.create([snapshot]) + + # Act + missing_intervals = snapshot.missing_intervals( + start="2020-01-01", end="2020-01-03", deployability_index=deployability_index + ) + + # Assert + assert not deployability_index.is_representative(snapshot) + assert missing_intervals == [ + (to_timestamp("2020-01-02"), to_timestamp("2020-01-03")), # Not missing from prod intervals + (to_timestamp("2020-01-03"), to_timestamp("2020-01-04")), + ] + + +def test__missing_intervals__representative_but_non_deployable_snapshot_no_intervals_returned( + snapshot: Snapshot, +): + # Arrange + snapshot.add_interval(start="2020-01-01", end="2020-01-01", is_dev=False) + snapshot.add_interval(start="2020-01-02", end="2020-01-02", is_dev=False) + snapshot.add_interval(start="2020-01-01", end="2020-01-01", is_dev=True) + + snapshot.categorize_as(SnapshotChangeCategory.INDIRECT_NON_BREAKING) + deployability_index = DeployabilityIndex.create([snapshot]) + + # Act + missing_intervals = snapshot.missing_intervals( + start="2020-01-01", end="2020-01-03", deployability_index=deployability_index + ) + + # Assert + assert deployability_index.is_representative(snapshot) + assert not deployability_index.is_deployable(snapshot) + assert missing_intervals == [] + + def test_incremental_time_self_reference(make_snapshot): snapshot = make_snapshot( SqlModel( @@ -2407,63 +2472,101 @@ def test_earliest_start_date(sushi_context: Context): def test_deployability_index(make_snapshot): - snapshot_a = make_snapshot(SqlModel(name="a", query=parse_one("SELECT 1"))) - snapshot_a.categorize_as(SnapshotChangeCategory.BREAKING) + # Breaking change - should be both deployable / representative + snapshot_breaking = make_snapshot(SqlModel(name="breaking", query=parse_one("SELECT 1"))) + snapshot_breaking.categorize_as(SnapshotChangeCategory.BREAKING) - snapshot_b = make_snapshot(SqlModel(name="b", query=parse_one("SELECT 1"))) - snapshot_b.categorize_as(SnapshotChangeCategory.BREAKING, forward_only=True) - snapshot_b.parents = (snapshot_a.snapshot_id,) + # Forward only breaking change - cannot be deployable / representative due to forward only + snapshot_breaking_forward_only = make_snapshot( + SqlModel(name="forward_only", query=parse_one("SELECT 1")) + ) + snapshot_breaking_forward_only.categorize_as(SnapshotChangeCategory.BREAKING, forward_only=True) + snapshot_breaking_forward_only.parents = (snapshot_breaking.snapshot_id,) - snapshot_c = make_snapshot(SqlModel(name="c", query=parse_one("SELECT 1"))) - snapshot_c.categorize_as(SnapshotChangeCategory.INDIRECT_BREAKING) - snapshot_c.parents = (snapshot_b.snapshot_id,) + # Indirect breaking - cannot be deployable / representative due to forward only non-representative parent + snapshot_indirect_breaking_non_representative_parent = make_snapshot( + SqlModel(name="indirect_breaking_non_representative_parent", query=parse_one("SELECT 1")) + ) + snapshot_indirect_breaking_non_representative_parent.categorize_as( + SnapshotChangeCategory.INDIRECT_BREAKING + ) + snapshot_indirect_breaking_non_representative_parent.parents = ( + snapshot_breaking_forward_only.snapshot_id, + ) - snapshot_d = make_snapshot(SqlModel(name="d", query=parse_one("SELECT 1"))) - snapshot_d.categorize_as(SnapshotChangeCategory.INDIRECT_BREAKING) - snapshot_d.parents = (snapshot_b.snapshot_id, snapshot_a.snapshot_id) + # Indirect breaking - can be deployable / representative due to forward only representative parent + snapshot_indirect_breaking_representative_parent = make_snapshot( + SqlModel(name="indirect_breaking_representative_parent", query=parse_one("SELECT 1")) + ) + snapshot_indirect_breaking_representative_parent.categorize_as( + SnapshotChangeCategory.INDIRECT_BREAKING + ) + snapshot_indirect_breaking_representative_parent.parents = (snapshot_breaking.snapshot_id,) - snapshot_e = make_snapshot(SqlModel(name="e", query=parse_one("SELECT 1"))) - snapshot_e.categorize_as(SnapshotChangeCategory.NON_BREAKING) + # Non breaking - deployable / representative due to no breaking changes + snapshot_non_breaking = make_snapshot( + SqlModel(name="non_breaking", query=parse_one("SELECT 1")) + ) + snapshot_non_breaking.categorize_as(SnapshotChangeCategory.NON_BREAKING) - snapshot_f = make_snapshot(SqlModel(name="f", query=parse_one("SELECT 1"))) - snapshot_f.categorize_as(SnapshotChangeCategory.INDIRECT_BREAKING) - snapshot_f.parents = (snapshot_e.snapshot_id, snapshot_a.snapshot_id) + # Indirect non breaking - can be representative but not deployable + snapshot_indirect_non_breaking = make_snapshot( + SqlModel(name="indirect_non_breaking", query=parse_one("SELECT 1")) + ) + snapshot_indirect_non_breaking.intervals = [ + (to_timestamp("2023-01-01"), to_timestamp("2023-01-02")) + ] + snapshot_indirect_non_breaking.categorize_as(SnapshotChangeCategory.INDIRECT_NON_BREAKING) + snapshot_indirect_non_breaking.parents = (snapshot_non_breaking.snapshot_id,) - snapshot_g = make_snapshot(SqlModel(name="g", query=parse_one("SELECT 1"))) - snapshot_g.intervals = [(to_timestamp("2023-01-01"), to_timestamp("2023-01-02"))] - snapshot_g.categorize_as(SnapshotChangeCategory.INDIRECT_NON_BREAKING) - snapshot_g.parents = (snapshot_e.snapshot_id,) + # Breaking with non-representative parent - cannot be deployable due to non-representative parent + snapshot_breaking_non_representative_parent = make_snapshot( + SqlModel(name="breaking_non_deployable_parents", query=parse_one("SELECT 1")) + ) + snapshot_breaking_non_representative_parent.categorize_as(SnapshotChangeCategory.BREAKING) + snapshot_breaking_non_representative_parent.parents = ( + snapshot_breaking_forward_only.snapshot_id, + ) snapshots = { s.snapshot_id: s for s in [ - snapshot_a, - snapshot_b, - snapshot_c, - snapshot_d, - snapshot_e, - snapshot_f, - snapshot_g, + snapshot_breaking, + snapshot_breaking_forward_only, + snapshot_indirect_breaking_non_representative_parent, + snapshot_indirect_breaking_representative_parent, + snapshot_non_breaking, + snapshot_indirect_non_breaking, + snapshot_breaking_non_representative_parent, ] } deployability_index = DeployabilityIndex.create(snapshots) - assert deployability_index.is_deployable(snapshot_a) - assert deployability_index.is_deployable(snapshot_e) - assert deployability_index.is_deployable(snapshot_f) - assert not deployability_index.is_deployable(snapshot_g) - assert not deployability_index.is_deployable(snapshot_b) - assert not deployability_index.is_deployable(snapshot_c) - assert not deployability_index.is_deployable(snapshot_d) + assert deployability_index.is_deployable(snapshot_breaking) + assert deployability_index.is_representative(snapshot_breaking) - assert deployability_index.is_representative(snapshot_a) - assert deployability_index.is_representative(snapshot_e) - assert deployability_index.is_representative(snapshot_f) - assert deployability_index.is_representative(snapshot_g) - assert not deployability_index.is_representative(snapshot_b) - assert not deployability_index.is_representative(snapshot_c) - assert not deployability_index.is_representative(snapshot_d) + assert not deployability_index.is_deployable(snapshot_breaking_forward_only) + assert not deployability_index.is_representative(snapshot_breaking_forward_only) + + assert deployability_index.is_deployable(snapshot_non_breaking) + assert deployability_index.is_representative(snapshot_non_breaking) + + assert not deployability_index.is_deployable( + snapshot_indirect_breaking_non_representative_parent + ) + assert not deployability_index.is_representative( + snapshot_indirect_breaking_non_representative_parent + ) + + assert deployability_index.is_deployable(snapshot_indirect_breaking_representative_parent) + assert deployability_index.is_representative(snapshot_indirect_breaking_representative_parent) + + assert not deployability_index.is_deployable(snapshot_indirect_non_breaking) + assert deployability_index.is_representative(snapshot_indirect_non_breaking) + + assert not deployability_index.is_deployable(snapshot_breaking_non_representative_parent) + assert not deployability_index.is_representative(snapshot_breaking_non_representative_parent) all_deployable_index = deployability_index.all_deployable() assert all(all_deployable_index.is_deployable(s) for s in snapshots.values()) diff --git a/tests/core/test_snapshot_evaluator.py b/tests/core/test_snapshot_evaluator.py index 3950e0d53f..4c0f48efbf 100644 --- a/tests/core/test_snapshot_evaluator.py +++ b/tests/core/test_snapshot_evaluator.py @@ -307,7 +307,7 @@ def increment_stage_counter(evaluator) -> None: ) -def test_promote(mocker: MockerFixture, adapter_mock, make_snapshot): +def test_promote__representative__prod_physical(mocker: MockerFixture, adapter_mock, make_snapshot): evaluator = SnapshotEvaluator(adapter_mock) model = SqlModel( @@ -319,9 +319,15 @@ def test_promote(mocker: MockerFixture, adapter_mock, make_snapshot): snapshot = make_snapshot(model) snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + deployability_index = DeployabilityIndex.create([snapshot]) - evaluator.promote([snapshot], EnvironmentNamingInfo(name="test_env")) + evaluator.promote( + target_snapshots=[snapshot], + environment_naming_info=EnvironmentNamingInfo(name="test_env"), + deployability_index=deployability_index, + ) + assert deployability_index.is_representative(snapshot) adapter_mock.transaction.assert_called() adapter_mock.session.assert_called() adapter_mock.create_schema.assert_called_once_with(to_schema("test_schema__test_env")) @@ -336,6 +342,43 @@ def test_promote(mocker: MockerFixture, adapter_mock, make_snapshot): ) +def test_promote__non_representative__dev_physical( + mocker: MockerFixture, adapter_mock, make_snapshot +): + evaluator = SnapshotEvaluator(adapter_mock) + + model = SqlModel( + name="test_schema.test_model", + kind=IncrementalByTimeRangeKind(time_column="a"), + storage_format="parquet", + query=parse_one("SELECT a FROM tbl WHERE ds BETWEEN @start_ds and @end_ds"), + ) + + snapshot = make_snapshot(model) + snapshot.categorize_as(SnapshotChangeCategory.BREAKING, forward_only=True) + deployability_index = DeployabilityIndex.create([snapshot]) + + evaluator.promote( + target_snapshots=[snapshot], + environment_naming_info=EnvironmentNamingInfo(name="test_env"), + deployability_index=deployability_index, + ) + + assert not deployability_index.is_representative(snapshot) + adapter_mock.transaction.assert_called() + adapter_mock.session.assert_called() + adapter_mock.create_schema.assert_called_once_with(to_schema("test_schema__test_env")) + adapter_mock.create_view.assert_called_once_with( + "test_schema__test_env.test_model", + parse_one( + f"SELECT * FROM sqlmesh__test_schema.test_schema__test_model__{snapshot.version}__dev" + ), + table_description=None, + column_descriptions=None, + view_properties={}, + ) + + def test_demote(mocker: MockerFixture, adapter_mock, make_snapshot): evaluator = SnapshotEvaluator(adapter_mock)