diff --git a/sqlmesh/core/plan/stages.py b/sqlmesh/core/plan/stages.py index 729e1705b4..e0e3955d57 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,10 @@ 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 +643,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 not deployability_index.is_eligible_for_physical_layer(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/scheduler.py b/sqlmesh/core/scheduler.py index 7345d92888..6e2903c664 100644 --- a/sqlmesh/core/scheduler.py +++ b/sqlmesh/core/scheduler.py @@ -185,11 +185,20 @@ def merged_missing_intervals( ignore_cron=ignore_cron, end_bounded=end_bounded, ) + + deployability_index = deployability_index or DeployabilityIndex.all_deployable() + # Filtering snapshots after computing missing intervals because we need all snapshots in order # to correctly infer start dates. + # + # missing_intervals will already have removed any snapshots that are representative but not + # deployable, an additional check is included here as a safeguard. if selected_snapshots is not None: snapshots_to_intervals = { - s: i for s, i in snapshots_to_intervals.items() if s.name in selected_snapshots + s: i + for s, i in snapshots_to_intervals.items() + if s.name in selected_snapshots + and deployability_index.is_eligible_for_physical_layer(s) } return snapshots_to_intervals diff --git a/sqlmesh/core/snapshot/definition.py b/sqlmesh/core/snapshot/definition.py index 2d813a06b1..723884859a 100644 --- a/sqlmesh/core/snapshot/definition.py +++ b/sqlmesh/core/snapshot/definition.py @@ -1048,6 +1048,11 @@ 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 not deployability_index.is_eligible_for_physical_layer(self): + return [] + intervals = ( self.intervals if deployability_index.is_representative(self) else self.dev_intervals ) @@ -1610,10 +1615,9 @@ 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. + """Returns true if the deployable (non-dev) table of the given snapshot should be used for reading and table mapping. - Note, that deployable snapshots are also representative, but the reverse is not always true. + Note, that deployable snapshots are always representative, but the reverse is not always true. Unlike `is_deployable`, this variant also captures FORWARD_ONLY and INDIRECT_NON_BREAKING snapshots that are not deployable by their nature but are currently promoted in production. Therefore, it's safe to consider @@ -1629,6 +1633,23 @@ def is_representative(self, snapshot: SnapshotIdLike) -> bool: representative = snapshot_id in self.representative_shared_version_ids return representative or self.is_deployable(snapshot) + def is_eligible_for_physical_layer(self, snapshot: SnapshotIdLike) -> bool: + """Returns true if the snapshot eligible for the physical layer. This will be true unless it is is representative + (looks like prod) but not deployable (can never become prod), and is used for computing missing intervals and + physical layer creation. + + This will only return false for FORWARD_ONLY and INDIRECT_NON_BREAKING snapshots in non-production environments. + + Args: + snapshot: The snapshot to check. + + Returns: + True if the snapshot is representative and not deployable, False otherwise. + """ + return not self.is_representative(snapshot=snapshot) or self.is_deployable( + snapshot=snapshot + ) + def with_non_deployable(self, snapshot: SnapshotIdLike) -> DeployabilityIndex: """Creates a new index with the given snapshot marked as non-deployable.""" return self._add_snapshot(snapshot, False) @@ -1666,6 +1687,11 @@ def create( start: t.Optional[TimeLike] = None, # plan start start_override_per_model: t.Optional[t.Dict[str, datetime]] = None, ) -> DeployabilityIndex: + def _snapshot_is_representative(snapshot: Snapshot) -> bool: + return not snapshot.is_paused or ( + snapshot.is_indirect_non_breaking and bool(snapshot.intervals) + ) + if not isinstance(snapshots, dict): snapshots = {s.snapshot_id: s for s in snapshots} @@ -1721,9 +1747,7 @@ def create( # Similarly, if the model depends on past and the start date is not aligned with the # model's start, we should consider this snapshot non-deployable. this_deployable = False - if not snapshot.is_paused or ( - snapshot.is_indirect_non_breaking and snapshot.intervals - ): + if _snapshot_is_representative(snapshot): # This snapshot represents what's currently deployed in prod. representative_shared_version_ids.add(node) else: @@ -1731,7 +1755,7 @@ def create( children_deployable = False else: children_deployable = False - if not snapshots[node].is_paused: + if _snapshot_is_representative(snapshot): representative_shared_version_ids.add(node) deployability_mapping[node] = this_deployable @@ -2067,6 +2091,11 @@ 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 not deployability_index.is_eligible_for_physical_layer(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/integration/test_config.py b/tests/core/integration/test_config.py index 5d571cd7c5..4bc820629d 100644 --- a/tests/core/integration/test_config.py +++ b/tests/core/integration/test_config.py @@ -280,15 +280,23 @@ def get_prod_dev_views(metadata: DuckDBMetadata) -> t.Tuple[t.Set[exp.Table], t. dev_views = {x for x in views if x.catalog == "dev_catalog" if x.db in environments_schemas} return prod_views, dev_views - def get_default_catalog_and_non_tables( + def get_default_catalog_and_non_sql_objects( metadata: DuckDBMetadata, default_catalog: t.Optional[str] - ) -> t.Tuple[t.Set[exp.Table], t.Set[exp.Table]]: + ) -> t.Tuple[t.Set[exp.Table], t.Set[exp.Table], t.Set[exp.Table], t.Set[exp.Table]]: tables = metadata.qualified_tables + views = metadata.qualified_views user_default_tables = { x for x in tables if x.catalog == default_catalog and x.db != "sqlmesh" } + user_default_views = { + x + for x in views + if x.catalog == default_catalog + and x.db not in (list(environments_schemas) + ["sqlmesh"]) + } non_default_tables = {x for x in tables if x.catalog != default_catalog} - return user_default_tables, non_default_tables + non_default_views = {x for x in views if x.catalog != default_catalog} + return user_default_tables, user_default_views, non_default_tables, non_default_views context, plan = init_and_plan_context( "examples/sushi", config="environment_catalog_mapping_config" @@ -299,11 +307,16 @@ def get_default_catalog_and_non_tables( prod_views, dev_views = get_prod_dev_views(metadata) ( user_default_tables, + user_default_views, non_default_tables, - ) = get_default_catalog_and_non_tables(metadata, context.default_catalog) + non_default_views, + ) = get_default_catalog_and_non_sql_objects(metadata, context.default_catalog) assert len(prod_views) == 16 assert len(dev_views) == 0 - assert len(user_default_tables) == 15 + assert ( + len(user_default_tables) == 15 + ) # One of the snapshots (top_waiters) is a view as opposed to a table. + assert len(user_default_views) == 1 assert state_metadata.schemas == ["sqlmesh"] assert {x.sql() for x in state_metadata.qualified_tables}.issuperset( { @@ -317,12 +330,16 @@ def get_default_catalog_and_non_tables( prod_views, dev_views = get_prod_dev_views(metadata) ( user_default_tables, + user_default_views, non_default_tables, - ) = get_default_catalog_and_non_tables(metadata, context.default_catalog) + non_default_views, + ) = get_default_catalog_and_non_sql_objects(metadata, context.default_catalog) assert len(prod_views) == 16 assert len(dev_views) == 16 - assert len(user_default_tables) == 16 + assert len(user_default_tables) == 15 + assert len(user_default_views) == 1 assert len(non_default_tables) == 0 + assert len(non_default_views) == 32 assert state_metadata.schemas == ["sqlmesh"] assert {x.sql() for x in state_metadata.qualified_tables}.issuperset( { @@ -336,12 +353,16 @@ def get_default_catalog_and_non_tables( prod_views, dev_views = get_prod_dev_views(metadata) ( user_default_tables, + user_default_views, non_default_tables, - ) = get_default_catalog_and_non_tables(metadata, context.default_catalog) + non_default_views, + ) = get_default_catalog_and_non_sql_objects(metadata, context.default_catalog) assert len(prod_views) == 16 assert len(dev_views) == 32 - assert len(user_default_tables) == 16 + assert len(user_default_tables) == 15 + assert len(user_default_views) == 1 assert len(non_default_tables) == 0 + assert len(non_default_views) == 48 assert state_metadata.schemas == ["sqlmesh"] assert {x.sql() for x in state_metadata.qualified_tables}.issuperset( { @@ -356,12 +377,16 @@ def get_default_catalog_and_non_tables( prod_views, dev_views = get_prod_dev_views(metadata) ( user_default_tables, + user_default_views, non_default_tables, - ) = get_default_catalog_and_non_tables(metadata, context.default_catalog) + non_default_views, + ) = get_default_catalog_and_non_sql_objects(metadata, context.default_catalog) assert len(prod_views) == 16 assert len(dev_views) == 16 - assert len(user_default_tables) == 16 + assert len(user_default_tables) == 15 + assert len(user_default_views) == 1 assert len(non_default_tables) == 0 + assert len(non_default_views) == 32 assert state_metadata.schemas == ["sqlmesh"] assert {x.sql() for x in state_metadata.qualified_tables}.issuperset( { diff --git a/tests/core/integration/test_forward_only.py b/tests/core/integration/test_forward_only.py index 2dddf18efd..a9ba184fd2 100644 --- a/tests/core/integration/test_forward_only.py +++ b/tests/core/integration/test_forward_only.py @@ -62,14 +62,7 @@ def test_forward_only_plan_with_effective_date(context_fixture: Context, request plan = plan_builder.set_effective_from("2023-01-05").build() # Default start should be set to effective_from assert plan.missing_intervals == [ - SnapshotIntervals( - snapshot_id=top_waiters_snapshot.snapshot_id, - intervals=[ - (to_timestamp("2023-01-05"), to_timestamp("2023-01-06")), - (to_timestamp("2023-01-06"), to_timestamp("2023-01-07")), - (to_timestamp("2023-01-07"), to_timestamp("2023-01-08")), - ], - ), + # top_waiters_snapshot has no missing intervals as it is an INDIRECT_NON_BREAKING change. SnapshotIntervals( snapshot_id=snapshot.snapshot_id, intervals=[ @@ -83,13 +76,7 @@ def test_forward_only_plan_with_effective_date(context_fixture: Context, request plan = plan_builder.set_start("2023-01-06").build() # Start override should take precedence assert plan.missing_intervals == [ - SnapshotIntervals( - snapshot_id=top_waiters_snapshot.snapshot_id, - intervals=[ - (to_timestamp("2023-01-06"), to_timestamp("2023-01-07")), - (to_timestamp("2023-01-07"), to_timestamp("2023-01-08")), - ], - ), + # top_waiters_snapshot has no missing intervals as it is an INDIRECT_NON_BREAKING change. SnapshotIntervals( snapshot_id=snapshot.snapshot_id, intervals=[ @@ -103,13 +90,7 @@ def test_forward_only_plan_with_effective_date(context_fixture: Context, request # Start should remain unchanged assert plan.start == "2023-01-06" assert plan.missing_intervals == [ - SnapshotIntervals( - snapshot_id=top_waiters_snapshot.snapshot_id, - intervals=[ - (to_timestamp("2023-01-06"), to_timestamp("2023-01-07")), - (to_timestamp("2023-01-07"), to_timestamp("2023-01-08")), - ], - ), + # top_waiters_snapshot has no missing intervals as it is an INDIRECT_NON_BREAKING change. SnapshotIntervals( snapshot_id=snapshot.snapshot_id, intervals=[ @@ -133,6 +114,7 @@ def test_forward_only_plan_with_effective_date(context_fixture: Context, request # Make sure that the previously set effective_from is respected assert prod_plan.start == to_timestamp("2023-01-04") assert prod_plan.missing_intervals == [ + # top_waiters_snapshot has missing intervals in the production environment despite it being an INDIRECT_NON_BREAKING change. SnapshotIntervals( snapshot_id=top_waiters_snapshot.snapshot_id, intervals=[ @@ -208,18 +190,7 @@ def test_forward_only_model_regular_plan(init_and_plan_context: t.Callable): ) plan_builder.set_start("2023-01-06") assert plan_builder.build().missing_intervals == [ - SnapshotIntervals( - snapshot_id=top_waiters_snapshot.snapshot_id, - intervals=[ - (to_timestamp("2023-01-01"), to_timestamp("2023-01-02")), - (to_timestamp("2023-01-02"), to_timestamp("2023-01-03")), - (to_timestamp("2023-01-03"), to_timestamp("2023-01-04")), - (to_timestamp("2023-01-04"), to_timestamp("2023-01-05")), - (to_timestamp("2023-01-05"), to_timestamp("2023-01-06")), - (to_timestamp("2023-01-06"), to_timestamp("2023-01-07")), - (to_timestamp("2023-01-07"), to_timestamp("2023-01-08")), - ], - ), + # top_waiters_snapshot has no missing intervals as it is an INDIRECT_NON_BREAKING change. SnapshotIntervals( snapshot_id=snapshot.snapshot_id, intervals=[ @@ -232,18 +203,7 @@ def test_forward_only_model_regular_plan(init_and_plan_context: t.Callable): # Make sure that changed start is reflected in missing intervals plan_builder.set_start("2023-01-07") assert plan_builder.build().missing_intervals == [ - SnapshotIntervals( - snapshot_id=top_waiters_snapshot.snapshot_id, - intervals=[ - (to_timestamp("2023-01-01"), to_timestamp("2023-01-02")), - (to_timestamp("2023-01-02"), to_timestamp("2023-01-03")), - (to_timestamp("2023-01-03"), to_timestamp("2023-01-04")), - (to_timestamp("2023-01-04"), to_timestamp("2023-01-05")), - (to_timestamp("2023-01-05"), to_timestamp("2023-01-06")), - (to_timestamp("2023-01-06"), to_timestamp("2023-01-07")), - (to_timestamp("2023-01-07"), to_timestamp("2023-01-08")), - ], - ), + # top_waiters_snapshot has no missing intervals as it is an INDIRECT_NON_BREAKING change. SnapshotIntervals( snapshot_id=snapshot.snapshot_id, intervals=[ diff --git a/tests/core/test_plan_stages.py b/tests/core/test_plan_stages.py index eb3f965761..a6bb82e237 100644 --- a/tests/core/test_plan_stages.py +++ b/tests/core/test_plan_stages.py @@ -78,6 +78,34 @@ def snapshot_c(make_snapshot, snapshot_a: Snapshot) -> Snapshot: return snapshot +@pytest.fixture +def snapshot_d(make_snapshot, snapshot_a: Snapshot) -> Snapshot: + snapshot = make_snapshot( + SqlModel( + name="d", + query=parse_one("select * from a"), + kind=dict(name=ModelKindName.VIEW), + ), + nodes={'"a"': snapshot_a.model}, + ) + snapshot.categorize_as(SnapshotChangeCategory.NON_BREAKING) + return snapshot + + +@pytest.fixture +def snapshot_e(make_snapshot, snapshot_d: Snapshot) -> Snapshot: + snapshot = make_snapshot( + SqlModel( + name="e", + query=parse_one("select * from d"), + kind=dict(name=ModelKindName.VIEW), + ), + nodes={'"d"': snapshot_d.model}, + ) + snapshot.categorize_as(SnapshotChangeCategory.INDIRECT_NON_BREAKING) + return snapshot + + def test_build_plan_stages_basic( snapshot_a: Snapshot, snapshot_b: Snapshot, mocker: MockerFixture ) -> None: @@ -1688,6 +1716,87 @@ def test_build_plan_stages_indirect_non_breaking_view_migration( assert isinstance(stages[8], FinalizeEnvironmentStage) +def test_build_plan_stages_indirect_non_breaking_view_dev( + snapshot_d: Snapshot, snapshot_e: Snapshot, make_snapshot, mocker: MockerFixture +) -> None: + # Setup Snapshots + snapshot_e.intervals = [(to_timestamp("2023-01-01"), to_timestamp("2023-01-02"))] + + # Create environment + state_reader = mocker.Mock(spec=StateReader) + state_reader.get_snapshots.return_value = {} + existing_environment = Environment( + name="prod", + snapshots=[snapshot_d.table_info, snapshot_e.table_info], + start_at="2023-01-01", + end_at="2023-01-02", + plan_id="previous_plan", + previous_plan_id=None, + promoted_snapshot_ids=[snapshot_d.snapshot_id, snapshot_e.snapshot_id], + finalized_ts=to_timestamp("2023-01-02"), + ) + state_reader.get_environment.return_value = existing_environment + + environment = Environment( + name="dev", + snapshots=[snapshot_d.table_info, snapshot_e.table_info], + start_at="2023-01-01", + end_at="2023-01-02", + plan_id="test_plan", + previous_plan_id="previous_plan", + promoted_snapshot_ids=[snapshot_d.snapshot_id, snapshot_e.snapshot_id], + ) + + # Create evaluatable plan + plan = EvaluatablePlan( + start="2023-01-01", + end="2023-01-02", + new_snapshots=[snapshot_d, snapshot_e], + environment=environment, + no_gaps=False, + skip_backfill=False, + empty_backfill=False, + restatements={}, + restate_all_snapshots=False, + is_dev=True, + allow_destructive_models=set(), + allow_additive_models=set(), + forward_only=False, + end_bounded=False, + ensure_finalized_snapshots=False, + ignore_cron=False, + directly_modified_snapshots=[snapshot_d.snapshot_id], + indirectly_modified_snapshots={ + snapshot_d.name: [snapshot_e.snapshot_id], + }, + metadata_updated_snapshots=[], + removed_snapshots=[], + requires_backfill=True, + models_to_backfill=None, + execution_time="2023-01-02", + disabled_restatement_models=set(), + environment_statements=None, + user_provided_flags=None, + ) + + # Build plan stages + stages = build_plan_stages(plan, state_reader, None) + + # Verify stages + assert len(stages) == 6 + + assert isinstance(stages[0], CreateSnapshotRecordsStage) + assert isinstance(stages[1], PhysicalLayerSchemaCreationStage) + assert isinstance(stages[2], BackfillStage) + assert isinstance(stages[3], EnvironmentRecordUpdateStage) + assert isinstance(stages[4], VirtualLayerUpdateStage) + assert isinstance(stages[5], FinalizeEnvironmentStage) + + ## Verify that indirect non breaking change does not trigger a schema creation or backfill + assert snapshot_e not in stages[1].snapshots + assert snapshot_e not in stages[2].snapshot_to_intervals + + def test_build_plan_stages_virtual_environment_mode_filtering( make_snapshot, mocker: MockerFixture ) -> None: diff --git a/tests/core/test_snapshot.py b/tests/core/test_snapshot.py index 64bee7f472..2ec37c8a5a 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,71 +2472,128 @@ 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_eligible_for_physical_layer(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_eligible_for_physical_layer(snapshot_breaking_forward_only) + + assert deployability_index.is_deployable(snapshot_non_breaking) + assert deployability_index.is_representative(snapshot_non_breaking) + assert deployability_index.is_eligible_for_physical_layer(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_eligible_for_physical_layer( + 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 deployability_index.is_eligible_for_physical_layer( + 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_eligible_for_physical_layer(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) + assert deployability_index.is_eligible_for_physical_layer( + 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()) assert all(all_deployable_index.is_representative(s) for s in snapshots.values()) + assert all( + all_deployable_index.is_eligible_for_physical_layer(s) for s in snapshots.values() + ) none_deployable_index = deployability_index.none_deployable() assert all(not none_deployable_index.is_deployable(s) for s in snapshots.values()) assert all(not none_deployable_index.is_representative(s) for s in snapshots.values()) + assert all( + none_deployable_index.is_eligible_for_physical_layer(s) for s in snapshots.values() + ) def test_deployability_index_unpaused_forward_only(make_snapshot): @@ -2483,15 +2605,18 @@ def test_deployability_index_unpaused_forward_only(make_snapshot): snapshot_b.categorize_as(SnapshotChangeCategory.BREAKING) snapshot_b.parents = (snapshot_a.snapshot_id,) - deplyability_index = DeployabilityIndex.create( + deployability_index = DeployabilityIndex.create( {s.snapshot_id: s for s in [snapshot_a, snapshot_b]} ) - assert not deplyability_index.is_deployable(snapshot_a) - assert deplyability_index.is_deployable(snapshot_b) + assert not deployability_index.is_deployable(snapshot_a) + assert deployability_index.is_deployable(snapshot_b) - assert deplyability_index.is_representative(snapshot_a) - assert deplyability_index.is_representative(snapshot_b) + assert deployability_index.is_representative(snapshot_a) + assert deployability_index.is_representative(snapshot_b) + + assert not deployability_index.is_eligible_for_physical_layer(snapshot_a) + assert deployability_index.is_eligible_for_physical_layer(snapshot_b) def test_deployability_index_unpaused_auto_restatement(make_snapshot): @@ -2518,17 +2643,21 @@ def test_deployability_index_unpaused_auto_restatement(make_snapshot): snapshot_c.categorize_as(SnapshotChangeCategory.BREAKING) snapshot_c.parents = (snapshot_b.snapshot_id,) - deplyability_index = DeployabilityIndex.create( + deployability_index = DeployabilityIndex.create( {s.snapshot_id: s for s in [snapshot_a, snapshot_b, snapshot_c]} ) - assert not deplyability_index.is_deployable(snapshot_a) - assert not deplyability_index.is_deployable(snapshot_b) - assert not deplyability_index.is_deployable(snapshot_c) + assert not deployability_index.is_deployable(snapshot_a) + assert not deployability_index.is_deployable(snapshot_b) + assert not deployability_index.is_deployable(snapshot_c) + + assert deployability_index.is_representative(snapshot_a) + assert deployability_index.is_representative(snapshot_b) + assert not deployability_index.is_representative(snapshot_c) - assert deplyability_index.is_representative(snapshot_a) - assert deplyability_index.is_representative(snapshot_b) - assert not deplyability_index.is_representative(snapshot_c) + assert not deployability_index.is_eligible_for_physical_layer(snapshot_a) + assert not deployability_index.is_eligible_for_physical_layer(snapshot_b) + assert deployability_index.is_eligible_for_physical_layer(snapshot_c) def test_deployability_index_uncategorized_forward_only_model(make_snapshot): @@ -2557,6 +2686,9 @@ def test_deployability_index_uncategorized_forward_only_model(make_snapshot): assert not deployability_index.is_representative(snapshot_a) assert not deployability_index.is_representative(snapshot_b) + assert deployability_index.is_eligible_for_physical_layer(snapshot_a) + assert deployability_index.is_eligible_for_physical_layer(snapshot_b) + def test_deployability_index_auto_restatement_model(make_snapshot): model_a = SqlModel( @@ -2586,6 +2718,9 @@ def test_deployability_index_auto_restatement_model(make_snapshot): assert not deployability_index.is_representative(snapshot_a) assert not deployability_index.is_representative(snapshot_b) + assert deployability_index.is_eligible_for_physical_layer(snapshot_a) + assert deployability_index.is_eligible_for_physical_layer(snapshot_b) + def test_deployability_index_categorized_forward_only_model(make_snapshot): model_a = SqlModel( @@ -2615,6 +2750,9 @@ def test_deployability_index_categorized_forward_only_model(make_snapshot): assert not deployability_index.is_representative(snapshot_a) assert not deployability_index.is_representative(snapshot_b) + assert deployability_index.is_eligible_for_physical_layer(snapshot_a) + assert deployability_index.is_eligible_for_physical_layer(snapshot_b) + def test_deployability_index_missing_parent(make_snapshot): snapshot_a = make_snapshot(SqlModel(name="a", query=parse_one("SELECT 1"))) @@ -2624,10 +2762,10 @@ def test_deployability_index_missing_parent(make_snapshot): snapshot_b.categorize_as(SnapshotChangeCategory.BREAKING, forward_only=True) snapshot_b.parents = (snapshot_a.snapshot_id,) - deplyability_index = DeployabilityIndex.create({snapshot_b.snapshot_id: snapshot_b}) + deployability_index = DeployabilityIndex.create({snapshot_b.snapshot_id: snapshot_b}) - assert not deplyability_index.is_deployable(snapshot_b) - assert not deplyability_index.is_deployable(snapshot_a) + assert not deployability_index.is_deployable(snapshot_b) + assert not deployability_index.is_deployable(snapshot_a) @pytest.mark.parametrize( diff --git a/tests/core/test_snapshot_evaluator.py b/tests/core/test_snapshot_evaluator.py index 0955af331d..fd84629e84 100644 --- a/tests/core/test_snapshot_evaluator.py +++ b/tests/core/test_snapshot_evaluator.py @@ -424,7 +424,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( @@ -436,9 +436,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")) @@ -453,6 +459,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) diff --git a/tests/utils/test_concurrency.py b/tests/utils/test_concurrency.py index 0d45a375f1..04f9748fd7 100644 --- a/tests/utils/test_concurrency.py +++ b/tests/utils/test_concurrency.py @@ -168,7 +168,7 @@ def _run_bounded_python(program: str) -> None: env=env, capture_output=True, text=True, - timeout=8, + timeout=16, ) assert result.returncode == 0, f"{result.stdout}\n{result.stderr}"