Eliminate strict snapshot id matching (for now)

Summary:
We sometimes add fields to classes in the snapshot hierarchy.
This causes an issue when we load a historical snapshot, apply the
new default values (from added columns) and then recompute the snapshot
id. You end up with a different snapshot id then was persisted. This
ended up triggering this invariant, which did incorporate this subtley.

This diff just disables the invariant for now and adds a test.

A more "correct" fix is something more like https://dagster.phacility.com/D2845
which ends up overriding the snapshot ID when fetched from storage.
However this is a safer fix in the short term.

Test Plan: BK

Reviewers: prha, nate

Reviewed By: prha

Differential Revision: https://dagster.phacility.com/D2843

Author: schrockn

python_modules/dagster/dagster/core/snap/execution_plan_snapshot.py

@@ -23,9 +23,10 @@ def __init__(self, execution_plan_snapshot, pipeline_index):
 
         self._step_index = {step.key: step for step in self.execution_plan_snapshot.steps}
 
-        check.invariant(
-            execution_plan_snapshot.pipeline_snapshot_id == pipeline_index.pipeline_snapshot_id
-        )
+        # https://github.com/dagster-io/dagster/issues/2442
+        # check.invariant(
+        #     execution_plan_snapshot.pipeline_snapshot_id == pipeline_index.pipeline_snapshot_id
+        # )
 
     def has_step(self, key):
         check.str_param(key, 'key')

python_modules/dagster/dagster_tests/compat_tests/snapshot_0_7_9_shapshot_id_creation_change/sqlite/history/runs.db

@@ -0,0 +1,58 @@
+from dagster.core.instance import DagsterInstance, InstanceRef
+from dagster.core.snap import (
+    ExecutionPlanIndex,
+    PipelineIndex,
+    create_execution_plan_snapshot_id,
+    create_pipeline_snapshot_id,
+)
+from dagster.utils import file_relative_path
+from dagster.utils.test import restore_directory
+
+
+# a change of schema in the snapshot hierarchy caused hashes to be different
+# when snapshots reloaded
+def test_0_7_10_snapshot_id_change():
+    test_dir = file_relative_path(__file__, 'snapshot_0_7_9_shapshot_id_creation_change/sqlite')
+    with restore_directory(test_dir):
+
+        instance = DagsterInstance.from_ref(InstanceRef.from_dir(test_dir))
+        # run_id = 'e297fa70-49e8-43f8-abfe-1634f02644f6'
+
+        old_pipeline_snapshot_id = '88528edde2ed64da3c39cca0da8ba2f7586c1a5d'
+        old_execution_plan_snapshot_id = '2246f8e5a10d21e15fbfa3773d7b2d0bc1fa9d3d'
+
+        pipeline_snapshot = instance.get_pipeline_snapshot(old_pipeline_snapshot_id)
+        ep_snapshot = instance.get_execution_plan_snapshot(old_execution_plan_snapshot_id)
+
+        # It is the pipeline snapshot that changed
+        # Verify that snapshot ids are not equal
+        assert create_pipeline_snapshot_id(pipeline_snapshot) != old_pipeline_snapshot_id
+
+        assert create_execution_plan_snapshot_id(ep_snapshot) == old_execution_plan_snapshot_id
+
+        # This previously failed with a check error
+        assert ExecutionPlanIndex(ep_snapshot, PipelineIndex(pipeline_snapshot))
+
+
+# Scripts to create this (run against 0.7.9)
+#
+# from dagster import pipeline, solid, DagsterInstance, execute_pipeline
+# from dagster.core.snap.utils import create_snapshot_id
+#
+# from dagster.serdes import serialize_pp
+#
+# @solid
+# def noop_solid(_):
+#     pass
+#
+# @pipeline
+# def noop_pipeline():
+#     noop_solid()
+#
+# instance = DagsterInstance.get()
+#
+# result = execute_pipeline(noop_pipeline, instance=instance)
+#
+# run_id = result.run_id
+
+# print(serialize_pp(run))