From dde66c586466884fd8531e13ba75b837df494349 Mon Sep 17 00:00:00 2001
From: Zach Probst <Zach_Probst@intuit.com>
Date: Thu, 1 Aug 2024 15:16:14 -0700
Subject: [PATCH 01/11] Add Data Structure and Logic Supporting Squash

---
 nodestream/schema/migrations/migrations.py    | 124 ++++++-
 nodestream/schema/migrations/operations.py    | 246 ++++++++++++++
 .../schema/migrations/project_migrations.py   |   2 +-
 .../schema/migrations/state_providers.py      |   2 +-
 .../unit/schema/migrations/test_migrations.py |  77 ++++-
 .../unit/schema/migrations/test_operations.py | 311 ++++++++++++++++++
 6 files changed, 748 insertions(+), 14 deletions(-)

diff --git a/nodestream/schema/migrations/migrations.py b/nodestream/schema/migrations/migrations.py
index 619556677..2f6874143 100644
--- a/nodestream/schema/migrations/migrations.py
+++ b/nodestream/schema/migrations/migrations.py
@@ -1,4 +1,4 @@
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Dict, Iterable, List
 
@@ -16,12 +16,14 @@ class Migration(LoadsFromYamlFile, SavesToYamlFile):
     name: str
     operations: List[Operation]
     dependencies: List[str]
+    replaces: List[str] = field(default_factory=list)
 
     def to_file_data(self):
         return {
             "name": self.name,
             "operations": [operation.to_file_data() for operation in self.operations],
             "dependencies": self.dependencies,
+            "replaces": self.replaces,
         }
 
     @classmethod
@@ -33,17 +35,19 @@ def from_file_data(cls, file_data):
                 for operation_file_data in file_data["operations"]
             ],
             dependencies=file_data["dependencies"],
+            replaces=file_data.get("replaces", []),
         )
 
     @classmethod
     def describe_yaml_schema(cls):
-        from schema import Schema
+        from schema import Optional, Schema
 
         return Schema(
             {
                 "name": str,
                 "operations": [Operation.describe_yaml_schema()],
                 "dependencies": [str],
+                Optional("replaces"): [str],
             }
         )
 
@@ -74,6 +78,64 @@ def write_to_file_with_default_name(self, directory: Path) -> Path:
         self.write_to_file(path)
         return path
 
+    def is_squashed_migration(self) -> bool:
+        """Check if this migration is a squashed migration.
+
+        Returns:
+            True if this migration is a squashed migration, False otherwise.
+        """
+        return len(self.replaces) > 0
+
+    def replaces_migration(self, migration: "Migration") -> bool:
+        """Check if this migration replaces another migration.
+
+        Args:
+            migration: The migration to check if this migration replaces.
+
+        Returns:
+            True if this migration replaces the other migration, False otherwise.
+        """
+        return migration.name in self.replaces
+
+    @classmethod
+    def squash(
+        cls,
+        new_name: str,
+        migrations: Iterable["Migration"],
+        optimize_operations: bool = True,
+    ) -> "Migration":
+        """Make a new migration as the squashed for the given migrations.
+
+        Args:
+            new_name: The name of the new migration.
+            migrations: The migrations to squash.
+            optimize_operations: Whether to optimize the operations
+            before squashing.
+
+        Returns:
+            The new migration that is the effective squash of the
+            given migrations.
+        """
+        names_of_migrations_being_squashed = {m.name for m in migrations}
+        effective_operations = [o for m in migrations for o in m.operations]
+        dependencies = list(
+            {
+                d
+                for m in migrations
+                for d in m.dependencies
+                if d not in names_of_migrations_being_squashed
+            }
+        )
+        if optimize_operations:
+            effective_operations = Operation.optimize_operations(effective_operations)
+
+        return cls(
+            name=new_name,
+            operations=effective_operations,
+            replaces=list(names_of_migrations_being_squashed),
+            dependencies=dependencies,
+        )
+
 
 @dataclass(frozen=True, slots=True)
 class MigrationGraph:
@@ -98,20 +160,60 @@ def get_migration(self, name: str) -> Migration:
         """
         return self.migrations_by_name[name]
 
-    def get_ordered_migration_plan(self) -> List[Migration]:
+    def get_ordered_migration_plan(
+        self, completed_migrations: List[Migration]
+    ) -> List[Migration]:
+        completed_migration_names = {m.name for m in completed_migrations}
+        replacement_index = {r: m for m in self.all_migrations() for r in m.replaces}
         plan_order = []
 
-        for migration in self.all_migrations():
-            for required_migration in self.plan_to_execute_migration(migration):
-                if required_migration not in plan_order:
-                    plan_order.append(required_migration)
+        for migration in self.topological_order():
+            if migration.name in completed_migration_names:
+                continue
+
+            # If we are considering a migration that has been replaced,
+            # we _only_ want to add it if at least one of the migrations
+            # replaced by the replacing migration has not been completed.
+            # IN otherwords, if the changes from a squashed migration have
+            # have (at least partially) been applied, we don't want to add
+            # the squashed migration to the plan but will want to add the
+            # migrations that were replaced to the plan.
+            if (replacement := replacement_index.get(migration.name)) is not None:
+                print("was replaced", migration.name)
+                if not any(
+                    r in completed_migration_names for r in replacement.replaces
+                ):
+                    print("skipping", migration.name)
+                    continue
+
+            # Similarly, if we are looking at a squashed migration, we want to
+            # add it to the plan only if none of the migrations that it replaces
+            # have been completed.
+            if migration.is_squashed_migration():
+                print("squashed", migration.name)
+                if any(r in completed_migration_names for r in migration.replaces):
+                    print("skipping", migration.name)
+                    continue
+
+            # Now here we are in one of three stats:
+            #
+            # 1. The migraiton is some "regular" migration that has not been
+            #       completed.
+            # 2. The migration is a squashed migration that we want to add to
+            #       the plan.
+            # 3. The migration is a migration that has been replaced by a
+            #       squashed migration but we've determined that at least one
+            #       of the migrations that it replaces has not been completed.
+            #
+            # In all of these cases, we want to add the migration to the plan.
+            plan_order.append(migration)
 
         return plan_order
 
-    def _iterative_dfs_traversal(self, start_node: Migration) -> List[Migration]:
+    def _iterative_dfs_traversal(self, *start_node: Migration) -> List[Migration]:
         visited_order = []
         visited_set = set()
-        stack = [(start_node, False)]
+        stack = [(n, False) for n in start_node]
 
         while stack:
             node, processed = stack.pop()
@@ -129,8 +231,8 @@ def _iterative_dfs_traversal(self, start_node: Migration) -> List[Migration]:
 
         return visited_order
 
-    def plan_to_execute_migration(self, migration: Migration) -> List[Migration]:
-        return self._iterative_dfs_traversal(migration)
+    def topological_order(self):
+        return self._iterative_dfs_traversal(*self.get_leaf_migrations())
 
     def all_migrations(self) -> Iterable[Migration]:
         """Iterate over all migrations in the graph."""
diff --git a/nodestream/schema/migrations/operations.py b/nodestream/schema/migrations/operations.py
index ea050745e..44119cdcc 100644
--- a/nodestream/schema/migrations/operations.py
+++ b/nodestream/schema/migrations/operations.py
@@ -91,6 +91,62 @@ def to_file_data(self):
             "arguments": asdict(self),
         }
 
+    def reduce(self, other: "Operation") -> List["Operation"]:
+        """Reduce this operation with another operation.
+
+        This method should be implemented by subclasses to reduce this
+        operation with another operation. The result of this operation should
+        be a list of operations that are equivalent to applying this operation
+        and then the other operation. If the operations cannot be reduced, then
+        this method should return both operations in a list.
+        """
+        return [self] if self == other else [self, other]
+
+    @classmethod
+    def optimize(cls, operations: List["Operation"]) -> List["Operation"]:
+        """Optimize a list of operations.
+
+        This method should be implemented by subclasses to optimize a list of
+        operations. The result of this operation should be a list of operations
+        that are equivalent to applying all of the operations in the input list.
+        """
+        return OperationReducer(operations).reduce()
+
+
+class OperationReducer:
+    """A class that can reduce a list of operations."""
+
+    def __init__(self, operations: List[Operation]):
+        self.operations = operations
+
+    def reduce(self) -> List[Operation]:
+        """Reduce the list of operations.
+
+        Returns:
+            A list of operations that are equivalent to applying all of the
+            operations in the input list.
+        """
+        current = self.operations
+        while True:
+            reduced = self.reduce_once(current)
+            if reduced == current:
+                break
+            current = reduced
+        return current
+
+    def reduce_once(self, operations: List[Operation]) -> List[Operation]:
+        # We are going to go from back to front and
+        # reduce exactly one operation per iteration.
+        for i, operation in enumerate(reversed(operations)):
+            others = operations[: len(operations) - i - 1]
+            for other in reversed(others):
+                reduced = operation.reduce(other)
+                if len(reduced) < 2:
+                    unchanged = [o for o in operations if o not in (operation, other)]
+                    return unchanged + reduced
+
+        return operations
+
 
 @dataclass(frozen=True, slots=True)
 class CreateNodeType(Operation):
@@ -123,6 +179,11 @@ def as_node_type(self) -> GraphObjectSchema:
         schema.add_properties(self.properties)
         return schema
 
+    def reduce(self, other: Operation) -> List[Operation]:
+        if isinstance(other, DropNodeType) and other.name == self.name:
+            return []
+        return Operation.reduce(self, other)
+
 
 @dataclass(frozen=True, slots=True)
 class CreateRelationshipType(Operation):
@@ -155,6 +216,11 @@ def as_relationship_type(self) -> GraphObjectSchema:
         schema.add_properties(self.properties)
         return schema
 
+    def reduce(self, other: Operation) -> List[Operation]:
+        if isinstance(other, DropRelationshipType) and other.name == self.name:
+            return []
+        return Operation.reduce(self, other)
+
 
 @dataclass(frozen=True, slots=True)
 class DropNodeType(Operation):
@@ -179,6 +245,18 @@ def proposed_index_name(self) -> str:
     def describe(self) -> str:
         return f"Drop node type {self.name}"
 
+    def reduce(self, other: Operation) -> List[Operation]:
+        if isinstance(other, CreateNodeType) and other.name == self.name:
+            return []
+
+        if (
+            isinstance(other, (AddNodeProperty, DropNodeProperty, RenameNodeProperty))
+            and other.node_type == self.name
+        ):
+            return [self]
+
+        return Operation.reduce(self, other)
+
 
 @dataclass(frozen=True, slots=True)
 class DropRelationshipType(Operation):
@@ -197,6 +275,25 @@ class DropRelationshipType(Operation):
     def describe(self) -> str:
         return f"Drop relationship {self.name}"
 
+    def reduce(self, other: Operation) -> List[Operation]:
+        if isinstance(other, CreateRelationshipType) and other.name == self.name:
+            return []
+
+        if (
+            isinstance(
+                other,
+                (
+                    AddRelationshipProperty,
+                    DropRelationshipProperty,
+                    RenameRelationshipProperty,
+                ),
+            )
+            and other.relationship_type == self.name
+        ):
+            return [self]
+
+        return Operation.reduce(self, other)
+
 
 @dataclass(frozen=True, slots=True)
 class RenameNodeProperty(Operation):
@@ -219,6 +316,26 @@ class RenameNodeProperty(Operation):
     def describe(self) -> str:
         return f"Rename node property {self.old_property_name} to {self.new_property_name} on node type {self.node_type}"
 
+    def reduce(self, other: Operation) -> List[Operation]:
+        # If we add and then rename, we can just add with the new name.
+        # if we rename and then drop, we can just drop the old name.
+        if (
+            isinstance(other, AddNodeProperty)
+            and other.node_type == self.node_type
+            and other.property_name == self.old_property_name
+        ):
+            return [
+                AddNodeProperty(self.node_type, self.new_property_name, other.default)
+            ]
+        if (
+            isinstance(other, DropNodeProperty)
+            and other.node_type == self.node_type
+            and other.property_name == self.old_property_name
+        ):
+            return [DropNodeProperty(self.node_type, self.old_property_name)]
+
+        return Operation.reduce(self, other)
+
 
 @dataclass(frozen=True, slots=True)
 class RenameRelationshipProperty(Operation):
@@ -241,6 +358,31 @@ class RenameRelationshipProperty(Operation):
     def describe(self) -> str:
         return f"Rename relationship property {self.old_property_name} to {self.new_property_name} on relationship type {self.relationship_type}"
 
+    def reduce(self, other: Operation) -> List[Operation]:
+        # If we add and then rename, we can just add with the new name.
+        # if we rename and then drop, we can just drop the old name.
+        if (
+            isinstance(other, AddRelationshipProperty)
+            and other.relationship_type == self.relationship_type
+            and other.property_name == self.old_property_name
+        ):
+            return [
+                AddRelationshipProperty(
+                    self.relationship_type, self.new_property_name, other.default
+                )
+            ]
+
+        if (
+            isinstance(other, DropRelationshipProperty)
+            and other.relationship_type == self.relationship_type
+            and other.property_name == self.old_property_name
+        ):
+            return [
+                DropRelationshipProperty(self.relationship_type, self.old_property_name)
+            ]
+
+        return Operation.reduce(self, other)
+
 
 @dataclass(frozen=True, slots=True)
 class RenameNodeType(Operation):
@@ -269,6 +411,17 @@ def new_proposed_index_name(self) -> str:
     def describe(self) -> str:
         return f"Rename node type {self.old_type} to {self.new_type}"
 
+    def reduce(self, other: Operation) -> List[Operation]:
+        # If we create and then rename, we can just create with the new name.
+        # If we rename and then drop, we can just drop the old name.
+        if isinstance(other, CreateNodeType) and other.name == self.old_type:
+            return [CreateNodeType(self.new_type, other.keys, other.properties)]
+
+        if isinstance(other, DropNodeType) and other.name == self.new_type:
+            return [DropNodeType(self.old_type)]
+
+        return Operation.reduce(self, other)
+
 
 @dataclass(frozen=True, slots=True)
 class RenameRelationshipType(Operation):
@@ -289,6 +442,17 @@ class RenameRelationshipType(Operation):
     def describe(self) -> str:
         return f"Rename relationship type {self.old_type} to {self.new_type}"
 
+    def reduce(self, other: Operation) -> List[Operation]:
+        # If we create and then rename, we can just create with the new name.
+        # If we rename and then drop, we can just drop the old name.
+        if isinstance(other, CreateRelationshipType) and other.name == self.old_type:
+            return [CreateRelationshipType(self.new_type, other.keys, other.properties)]
+
+        if isinstance(other, DropRelationshipType) and other.name == self.new_type:
+            return [DropRelationshipType(self.old_type)]
+
+        return Operation.reduce(self, other)
+
 
 @dataclass(frozen=True, slots=True)
 class AddAdditionalNodePropertyIndex(Operation):
@@ -400,6 +564,27 @@ class AddNodeProperty(Operation):
     def describe(self) -> str:
         return f"Add property {self.property_name} to node type {self.node_type}"
 
+    def reduce(self, other: Operation) -> List[Operation]:
+        # If we add and then rename, we can just add with the new name.
+        # If we add and then drop, we can just no-op.
+        if (
+            isinstance(other, RenameNodeProperty)
+            and other.node_type == self.node_type
+            and other.old_property_name == self.property_name
+        ):
+            return [
+                AddNodeProperty(self.node_type, other.new_property_name, self.default)
+            ]
+
+        if (
+            isinstance(other, DropNodeProperty)
+            and other.node_type == self.node_type
+            and other.property_name == self.property_name
+        ):
+            return []
+
+        return Operation.reduce(self, other)
+
 
 @dataclass(frozen=True, slots=True)
 class AddRelationshipProperty(Operation):
@@ -423,6 +608,29 @@ class AddRelationshipProperty(Operation):
     def describe(self) -> str:
         return f"Add property {self.property_name} to relationship type {self.relationship_type}"
 
+    def reduce(self, other: Operation) -> List[Operation]:
+        # If we add and then rename, we can just add with the new name.
+        # If we add and then drop, we can just no-op.
+        if (
+            isinstance(other, RenameRelationshipProperty)
+            and other.relationship_type == self.relationship_type
+            and other.old_property_name == self.property_name
+        ):
+            return [
+                AddRelationshipProperty(
+                    self.relationship_type, other.new_property_name, self.default
+                )
+            ]
+
+        if (
+            isinstance(other, DropRelationshipProperty)
+            and other.relationship_type == self.relationship_type
+            and other.property_name == self.property_name
+        ):
+            return []
+
+        return Operation.reduce(self, other)
+
 
 @dataclass(frozen=True, slots=True)
 class DropNodeProperty(Operation):
@@ -442,6 +650,25 @@ class DropNodeProperty(Operation):
     def describe(self) -> str:
         return f"Drop property {self.property_name} from node type {self.node_type}"
 
+    def reduce(self, other: Operation) -> List[Operation]:
+        # If we add and then drop, we can just no-op.
+        # If we rename and then drop, we can just drop the original name.
+        if (
+            isinstance(other, AddNodeProperty)
+            and other.node_type == self.node_type
+            and other.property_name == self.property_name
+        ):
+            return []
+
+        if (
+            isinstance(other, RenameNodeProperty)
+            and other.node_type == self.node_type
+            and other.old_property_name == self.property_name
+        ):
+            return [self]
+
+        return Operation.reduce(self, other)
+
 
 @dataclass(frozen=True, slots=True)
 class DropRelationshipProperty(Operation):
@@ -461,6 +688,25 @@ class DropRelationshipProperty(Operation):
     def describe(self) -> str:
         return f"Drop property {self.property_name} from relationship type {self.relationship_type}"
 
+    def reduce(self, other: Operation) -> List[Operation]:
+        # If we add and then drop, we can just no-op.
+        # If we rename and then drop, we can just drop the original name.
+        if (
+            isinstance(other, AddRelationshipProperty)
+            and other.relationship_type == self.relationship_type
+            and other.property_name == self.property_name
+        ):
+            return []
+
+        if (
+            isinstance(other, RenameRelationshipProperty)
+            and other.relationship_type == self.relationship_type
+            and other.old_property_name == self.property_name
+        ):
+            return [self]
+
+        return Operation.reduce(self, other)
+
 
 @dataclass(frozen=True, slots=True)
 class NodeKeyExtended(Operation):
diff --git a/nodestream/schema/migrations/project_migrations.py b/nodestream/schema/migrations/project_migrations.py
index 65411b842..59fffe3c5 100644
--- a/nodestream/schema/migrations/project_migrations.py
+++ b/nodestream/schema/migrations/project_migrations.py
@@ -49,7 +49,7 @@ async def determine_pending(
             Each migration and a boolean indicating if it is pending.
         """
         completed_migrations = await migrator.get_completed_migrations(self.graph)
-        for migration in self.graph.get_ordered_migration_plan():
+        for migration in self.graph.topological_order():
             yield migration, migration not in completed_migrations
 
     async def execute_pending(self, migrator: Migrator) -> AsyncIterable[Migration]:
diff --git a/nodestream/schema/migrations/state_providers.py b/nodestream/schema/migrations/state_providers.py
index 20fd9153e..e37fb4cf3 100644
--- a/nodestream/schema/migrations/state_providers.py
+++ b/nodestream/schema/migrations/state_providers.py
@@ -51,7 +51,7 @@ def __init__(self, migrations: MigrationGraph) -> None:
 
     async def get_schema(self) -> Schema:
         migrator = InMemoryMigrator()
-        for migration in self.migrations.get_ordered_migration_plan():
+        for migration in self.migrations.get_ordered_migration_plan([]):
             await migrator.execute_migration(migration)
         return migrator.schema
 
diff --git a/tests/unit/schema/migrations/test_migrations.py b/tests/unit/schema/migrations/test_migrations.py
index 0a6161154..30f52ceb5 100644
--- a/tests/unit/schema/migrations/test_migrations.py
+++ b/tests/unit/schema/migrations/test_migrations.py
@@ -66,7 +66,7 @@ def test_migration_to_and_from_file_data():
     ],
 )
 def test_migration_graph_test_get_ordered_migration_plan(migration_graph):
-    plan = migration_graph.get_ordered_migration_plan()
+    plan = migration_graph.get_ordered_migration_plan([])
     assert_that(len(plan), is_(equal_to(len(migration_graph.migrations_by_name))))
     for i, migration in enumerate(plan):
         migrations_to_current = {m.name for m in plan[0:i]}
@@ -110,3 +110,78 @@ def test_migration_graph_from_directory(tmp_path):
 def test_migration_graph_get_by_name(migration_graph, root_migration):
     result = migration_graph.get_migration("root_migration")
     assert_that(result, equal_to(root_migration))
+
+
+def test_migration_graph_navigates_around_squashed_migrations_when_partially_applied():
+    a = Migration(name="a", operations=[], dependencies=[])
+    b = Migration(name="b", operations=[], dependencies=["a"])
+    c = Migration(name="c", operations=[], dependencies=["b"])
+    squash = Migration(
+        name="squash", operations=[], dependencies=["a"], replaces=["b", "c"]
+    )
+    d = Migration(name="d", operations=[], dependencies=["squash"])
+
+    graph = MigrationGraph.from_iterable([a, b, c, squash, d])
+
+    plan = graph.get_ordered_migration_plan([a, b])
+    assert_that(plan, is_(equal_to([d, c])))
+
+
+def test_migration_graph_navigates_around_squashed_migrations_when_fully_applied():
+    a = Migration(name="a", operations=[], dependencies=[])
+    b = Migration(name="b", operations=[], dependencies=["a"])
+    c = Migration(name="c", operations=[], dependencies=["b"])
+    squash = Migration(
+        name="squash", operations=[], dependencies=["a"], replaces=["b", "c"]
+    )
+    d = Migration(name="d", operations=[], dependencies=["squash"])
+
+    graph = MigrationGraph.from_iterable([a, b, c, squash, d])
+
+    plan = graph.get_ordered_migration_plan([a, b, c])
+    assert_that(plan, is_(equal_to([d])))
+
+
+def test_migration_graph_navigates_around_squashed_migrations_when_nothing_applied():
+    a = Migration(name="a", operations=[], dependencies=[])
+    b = Migration(name="b", operations=[], dependencies=["a"])
+    c = Migration(name="c", operations=[], dependencies=["b"])
+    squash = Migration(
+        name="squash", operations=[], dependencies=["a"], replaces=["b", "c"]
+    )
+    d = Migration(name="d", operations=[], dependencies=["squash"])
+
+    graph = MigrationGraph.from_iterable([a, b, c, squash, d])
+
+    plan = graph.get_ordered_migration_plan([])
+    assert_that(plan, is_(equal_to([a, squash, d])))
+
+
+def test_migration_graph_navigates_around_squashed_migrations_when_nothing_applied():
+    a = Migration(name="a", operations=[], dependencies=[])
+    b = Migration(name="b", operations=[], dependencies=["a"])
+    c = Migration(name="c", operations=[], dependencies=["b"])
+    squash = Migration(
+        name="squash", operations=[], dependencies=["a"], replaces=["b", "c"]
+    )
+    d = Migration(name="d", operations=[], dependencies=["squash"])
+
+    graph = MigrationGraph.from_iterable([a, b, c, squash, d])
+
+    plan = graph.get_ordered_migration_plan([])
+    assert_that(plan, is_(equal_to([a, squash, d])))
+
+
+def test_migraiton_graph_handles_dependencies_fulfilled_through_squash():
+    a = Migration(name="a", operations=[], dependencies=[])
+    b = Migration(name="b", operations=[], dependencies=["a"])
+    c = Migration(name="c", operations=[], dependencies=["b"])
+    squash = Migration(
+        name="squash", operations=[], dependencies=["a"], replaces=["b", "c"]
+    )
+    d = Migration(name="d", operations=[], dependencies=["b"])
+
+    graph = MigrationGraph.from_iterable([a, b, c, squash, d])
+
+    plan = graph.get_ordered_migration_plan([a, squash])
+    assert_that(plan, is_(equal_to([d])))
diff --git a/tests/unit/schema/migrations/test_operations.py b/tests/unit/schema/migrations/test_operations.py
index db4cf31af..e4996019e 100644
--- a/tests/unit/schema/migrations/test_operations.py
+++ b/tests/unit/schema/migrations/test_operations.py
@@ -303,3 +303,314 @@ def test_rename_node_type_new_name():
 def test_drop_node_type_index_name():
     name = DropNodeType("Person").proposed_index_name
     assert_that(name, equal_to("Person_node_key"))
+
+
+def test_create_node_type_reduce():
+    a = CreateNodeType("Person", {"name"}, {"age"})
+    b = DropNodeType("Person")
+    assert_that(a.reduce(b), equal_to([]))
+
+
+def test_create_node_type_reduce_no_match():
+    a = CreateNodeType("Person", {"name"}, {"age"})
+    b = DropNodeType("Human")
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_create_node_type_reduce_irrelevant_type():
+    a = CreateNodeType("Person", {"name"}, {"age"})
+    b = AddNodeProperty("Person", "full_name")
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_create_relationship_type_reduce():
+    a = CreateRelationshipType("KNOWS", {"since"}, {"since"})
+    b = DropRelationshipType("KNOWS")
+    assert_that(a.reduce(b), equal_to([]))
+
+
+def test_create_relationship_type_reduce_no_match():
+    a = CreateRelationshipType("KNOWS", {"since"}, {"since"})
+    b = DropRelationshipType("LIKES")
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_create_relationship_type_reduce_irrelevant_type():
+    a = CreateRelationshipType("KNOWS", {"since"}, {"since"})
+    b = AddRelationshipProperty("KNOWS", "known_since")
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_drop_node_type_reduce():
+    a = DropNodeType("Person")
+    b = CreateNodeType("Person", {"name"}, {"age"})
+    assert_that(a.reduce(b), equal_to([]))
+
+
+def test_drop_node_type_reduce_no_match():
+    a = DropNodeType("Person")
+    b = CreateNodeType("Human", {"name"}, {"age"})
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_drop_node_type_reduce_irrelevant_type():
+    a = DropNodeType("Person")
+    b = AddNodeProperty("Person", "full_name")
+    assert_that(a.reduce(b), equal_to([a]))
+
+
+def test_drop_relationship_type_reduce():
+    a = DropRelationshipType("KNOWS")
+    b = CreateRelationshipType("KNOWS", {"since"}, {"since"})
+    assert_that(a.reduce(b), equal_to([]))
+
+
+def test_drop_relationship_type_reduce_no_match():
+    a = DropRelationshipType("KNOWS")
+    b = CreateRelationshipType("LIKES", {"since"}, {"since"})
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_drop_relationship_type_reduce_dropping_added_property():
+    a = DropRelationshipType("KNOWS")
+    b = AddRelationshipProperty("KNOWS", "known_since")
+    assert_that(a.reduce(b), equal_to([a]))
+
+
+def test_rename_node_property_reduce_add():
+    a = RenameNodeProperty("Person", "full_name", "fuller_name")
+    b = AddNodeProperty("Person", "full_name")
+    assert_that(a.reduce(b), equal_to([AddNodeProperty("Person", "fuller_name")]))
+
+
+def test_rename_node_property_reduce_drop():
+    a = RenameNodeProperty("Person", "full_name", "fuller_name")
+    b = DropNodeProperty("Person", "full_name")
+    assert_that(a.reduce(b), equal_to([DropNodeProperty("Person", "full_name")]))
+
+
+def test_rename_node_property_reduce_no_match():
+    a = RenameNodeProperty("Person", "full_name", "fuller_name")
+    b = AddNodeProperty("Person", "name")
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_rename_node_property_reduce_irrelevant_type():
+    a = RenameNodeProperty("Person", "full_name", "fuller_name")
+    b = AddRelationshipProperty("KNOWS", "known_since")
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_rename_relationship_property_reduce_add():
+    a = RenameRelationshipProperty("KNOWS", "known_since", "known_since_date")
+    b = AddRelationshipProperty("KNOWS", "known_since")
+    assert_that(
+        a.reduce(b), equal_to([AddRelationshipProperty("KNOWS", "known_since_date")])
+    )
+
+
+def test_rename_relationship_property_reduce_drop():
+    a = RenameRelationshipProperty("KNOWS", "known_since", "known_since_date")
+    b = DropRelationshipProperty("KNOWS", "known_since")
+    assert_that(
+        a.reduce(b), equal_to([DropRelationshipProperty("KNOWS", "known_since")])
+    )
+
+
+def test_rename_relationship_property_reduce_no_match():
+    a = RenameRelationshipProperty("KNOWS", "known_since", "known_since_date")
+    b = AddRelationshipProperty("KNOWS", "since")
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_rename_relationship_property_reduce_irrelevant_type():
+    a = RenameRelationshipProperty("KNOWS", "known_since", "known_since_date")
+    b = AddNodeProperty("Person", "full_name")
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_rename_node_type_reduce():
+    a = RenameNodeType("Person", "Human")
+    b = CreateNodeType("Person", {"name"}, {"age"})
+    assert_that(a.reduce(b), equal_to([CreateNodeType("Human", {"name"}, {"age"})]))
+
+
+def test_rename_node_type_reduce_drop():
+    a = RenameNodeType("Person", "Human")
+    b = DropNodeType("Human")
+    assert_that(a.reduce(b), equal_to([DropNodeType("Person")]))
+
+
+def test_rename_node_type_reduce_no_match():
+    a = RenameNodeType("Person", "Human")
+    b = CreateNodeType("Human", {"name"}, {"age"})
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_rename_node_type_reduce_irrelevant_type():
+    a = RenameNodeType("Person", "Human")
+    b = AddNodeProperty("Person", "full_name")
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_rename_relationship_type_reduce():
+    a = RenameRelationshipType("KNOWS", "LIKES")
+    b = CreateRelationshipType("KNOWS", {"since"}, {"since"})
+    assert_that(
+        a.reduce(b), equal_to([CreateRelationshipType("LIKES", {"since"}, {"since"})])
+    )
+
+
+def test_rename_relationship_type_reduce_drop():
+    a = RenameRelationshipType("KNOWS", "LIKES")
+    b = DropRelationshipType("LIKES")
+    assert_that(a.reduce(b), equal_to([DropRelationshipType("KNOWS")]))
+
+
+def test_rename_relationship_type_reduce_no_match():
+    a = RenameRelationshipType("KNOWS", "LIKES")
+    b = CreateRelationshipType("LIKES", {"since"}, {"since"})
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_rename_relationship_type_reduce_irrelevant_type():
+    a = RenameRelationshipType("KNOWS", "LIKES")
+    b = AddNodeProperty("Person", "full_name")
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_add_node_property_reduce_drop():
+    a = AddNodeProperty("Person", "full_name")
+    b = DropNodeProperty("Person", "full_name")
+    assert_that(a.reduce(b), equal_to([]))
+
+
+def test_add_node_property_reduce_rename():
+    a = AddNodeProperty("Person", "full_name")
+    b = RenameNodeProperty("Person", "full_name", "fuller_name")
+    assert_that(a.reduce(b), equal_to([AddNodeProperty("Person", "fuller_name")]))
+
+
+def test_add_node_property_reduce_no_match():
+    a = AddNodeProperty("Person", "full_name")
+    b = DropNodeProperty("Person", "name")
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_add_node_property_reduce_irrelevant_type():
+    a = AddNodeProperty("Person", "full_name")
+    b = AddRelationshipProperty("KNOWS", "known_since")
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_add_relationship_property_reduce_drop():
+    a = AddRelationshipProperty("KNOWS", "known_since")
+    b = DropRelationshipProperty("KNOWS", "known_since")
+    assert_that(a.reduce(b), equal_to([]))
+
+
+def test_add_relationship_property_reduce_rename():
+    a = AddRelationshipProperty("KNOWS", "known_since")
+    b = RenameRelationshipProperty("KNOWS", "known_since", "known_since_date")
+    assert_that(
+        a.reduce(b), equal_to([AddRelationshipProperty("KNOWS", "known_since_date")])
+    )
+
+
+def test_add_relationship_property_reduce_no_match():
+    a = AddRelationshipProperty("KNOWS", "known_since")
+    b = DropRelationshipProperty("KNOWS", "since")
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_add_relationship_property_reduce_irrelevant_type():
+    a = AddRelationshipProperty("KNOWS", "known_since")
+    b = AddNodeProperty("Person", "full_name")
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_drop_node_property_reduce():
+    a = DropNodeProperty("Person", "full_name")
+    b = AddNodeProperty("Person", "full_name")
+    assert_that(a.reduce(b), equal_to([]))
+
+
+def test_drop_node_property_reduce_rename():
+    a = DropNodeProperty("Person", "full_name")
+    b = RenameNodeProperty("Person", "full_name", "fuller_name")
+    assert_that(a.reduce(b), equal_to([DropNodeProperty("Person", "full_name")]))
+
+
+def test_drop_node_property_reduce_no_match():
+    a = DropNodeProperty("Person", "full_name")
+    b = AddNodeProperty("Person", "name")
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_drop_node_property_reduce_irrelevant_type():
+    a = DropNodeProperty("Person", "full_name")
+    b = AddRelationshipProperty("KNOWS", "known_since")
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_drop_relationship_property_reduce():
+    a = DropRelationshipProperty("KNOWS", "known_since")
+    b = AddRelationshipProperty("KNOWS", "known_since")
+    assert_that(a.reduce(b), equal_to([]))
+
+
+def test_drop_relationship_property_reduce_rename():
+    a = DropRelationshipProperty("KNOWS", "known_since")
+    b = RenameRelationshipProperty("KNOWS", "known_since", "known_since_date")
+    assert_that(
+        a.reduce(b), equal_to([DropRelationshipProperty("KNOWS", "known_since")])
+    )
+
+
+def test_drop_relationship_property_reduce_no_match():
+    a = DropRelationshipProperty("KNOWS", "known_since")
+    b = AddRelationshipProperty("KNOWS", "since")
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_drop_relationship_property_reduce_irrelevant_type():
+    a = DropRelationshipProperty("KNOWS", "known_since")
+    b = AddNodeProperty("Person", "full_name")
+    assert_that(a.reduce(b), equal_to([a, b]))
+
+
+def test_operation_chain_optimization():
+    a = CreateNodeType("Person", {"name"}, {"age"})
+    b = CreateNodeType("Sport", {"name"}, {})
+    c = AddNodeProperty("Person", "full_name")
+    d = DropNodeProperty("Person", "age")
+    e = DropNodeType("Person")
+
+    assert_that(
+        Operation.optimize([a, b, c, d, e]),
+        equal_to([CreateNodeType("Sport", {"name"}, {})]),
+    )
+
+
+def test_operation_chain_optimization_complex():
+    a = CreateNodeType("Person", {"name"}, {"age"})
+    b = CreateNodeType("Sport", {"name"}, {})
+    c = AddNodeProperty("Person", "full_name")
+    d = DropNodeProperty("Person", "age")
+    e = DropNodeType("Person")
+    f = CreateRelationshipType("Likes", {}, {})
+    g = AddRelationshipProperty("Likes", "since")
+    h = AddNodeProperty("Team", "mascot")
+    i = DropRelationshipType("Likes")
+    j = CreateNodeType("Team", {"name"}, {})
+    k = DropNodeProperty("Team", "mascot")
+    l = DropRelationshipProperty("Likes", "since")
+    m = DropNodeType("Team")
+
+    # Intertwined and shuffled operations
+    operations = [a, f, b, g, c, j, h, d, k, i, e, l, m]
+
+    assert_that(
+        Operation.optimize(operations),
+        equal_to([CreateNodeType("Sport", {"name"}, {})]),
+    )

From 5d782c0818ce89b1ba4b8b7bb970dba1e6b234ed Mon Sep 17 00:00:00 2001
From: Zach Probst <Zach_Probst@intuit.com>
Date: Thu, 1 Aug 2024 15:52:44 -0700
Subject: [PATCH 02/11] Add higher order and CLI APIs

---
 nodestream/cli/commands/__init__.py           |  2 ++
 nodestream/cli/commands/squash_migrations.py  | 34 +++++++++++++++++++
 nodestream/cli/operations/__init__.py         |  2 ++
 .../operations/generate_squashed_migration.py | 29 ++++++++++++++++
 nodestream/schema/migrations/migrations.py    | 22 ++++++++++--
 .../schema/migrations/project_migrations.py   | 17 ++++++++++
 .../cli/commands/test_squash_migrations.py    | 17 ++++++++++
 .../test_generate_squashed_migration.py       | 22 ++++++++++++
 8 files changed, 142 insertions(+), 3 deletions(-)
 create mode 100644 nodestream/cli/commands/squash_migrations.py
 create mode 100644 nodestream/cli/operations/generate_squashed_migration.py
 create mode 100644 tests/unit/cli/commands/test_squash_migrations.py
 create mode 100644 tests/unit/cli/operations/test_generate_squashed_migration.py

diff --git a/nodestream/cli/commands/__init__.py b/nodestream/cli/commands/__init__.py
index 310cd74e5..2b4a5cb34 100644
--- a/nodestream/cli/commands/__init__.py
+++ b/nodestream/cli/commands/__init__.py
@@ -10,6 +10,7 @@
 from .scaffold import Scaffold
 from .show import Show
 from .show_migrations import ShowMigrations
+from .squash_migrations import SquashMigration
 
 __all__ = (
     "AuditCommand",
@@ -24,4 +25,5 @@
     "Scaffold",
     "ShowMigrations",
     "Show",
+    "SquashMigration",
 )
diff --git a/nodestream/cli/commands/squash_migrations.py b/nodestream/cli/commands/squash_migrations.py
new file mode 100644
index 000000000..e468c2fc8
--- /dev/null
+++ b/nodestream/cli/commands/squash_migrations.py
@@ -0,0 +1,34 @@
+from cleo.helpers import option
+
+from ..operations import GenerateSquashedMigration
+from .nodestream_command import NodestreamCommand
+from .shared_options import PROJECT_FILE_OPTION
+
+
+class SquashMigration(NodestreamCommand):
+    name = "migrations squash"
+    description = "Generate a migration for the current project."
+    options = [
+        PROJECT_FILE_OPTION,
+        option(
+            "from",
+            description="The name of the migration to squash from.",
+            value_required=True,
+        ),
+        option(
+            "to",
+            description="The name of the migration to squash to.",
+            value_required=True,
+        ),
+    ]
+
+    async def handle_async(self):
+        from_migration_name = self.option("from")
+        to_migration_name = self.option("to")
+        migrations = self.get_migrations()
+        operation = GenerateSquashedMigration(
+            migrations,
+            from_migration_name,
+            to_migration_name,
+        )
+        await self.run_operation(operation)
diff --git a/nodestream/cli/operations/__init__.py b/nodestream/cli/operations/__init__.py
index 9fbb77692..3a401bb74 100644
--- a/nodestream/cli/operations/__init__.py
+++ b/nodestream/cli/operations/__init__.py
@@ -3,6 +3,7 @@
 from .execute_migration import ExecuteMigrations
 from .generate_migration import GenerateMigration
 from .generate_pipeline_scaffold import GeneratePipelineScaffold
+from .generate_squashed_migration import GenerateSquashedMigration
 from .initialize_logger import InitializeLogger
 from .initialize_project import InitializeProject
 from .operation import Operation
@@ -20,6 +21,7 @@
     "ExecuteMigrations",
     "GenerateMigration",
     "GeneratePipelineScaffold",
+    "GenerateSquashedMigration",
     "InitializeLogger",
     "InitializeProject",
     "Operation",
diff --git a/nodestream/cli/operations/generate_squashed_migration.py b/nodestream/cli/operations/generate_squashed_migration.py
new file mode 100644
index 000000000..73dd810db
--- /dev/null
+++ b/nodestream/cli/operations/generate_squashed_migration.py
@@ -0,0 +1,29 @@
+from ...schema.migrations import ProjectMigrations
+from .operation import NodestreamCommand, Operation
+
+
+class GenerateSquashedMigration(Operation):
+    def __init__(
+        self,
+        migrations: ProjectMigrations,
+        from_migration_name: str,
+        to_migration_name: str,
+    ) -> None:
+        self.migrations = migrations
+        self.from_migration_name = from_migration_name
+        self.to_migration_name = to_migration_name
+
+    async def perform(self, command: NodestreamCommand):
+        from_migration = self.migrations.graph.get_migration(self.from_migration_name)
+        to_migration = self.migrations.graph.get_migration(self.to_migration_name)
+        migration, path = self.migrations.create_squash_between(
+            from_migration, to_migration
+        )
+        command.line(f"Generated squashed migration {migration.name}.")
+        command.line(
+            f"The migration contains {len(migration.operations)} schema changes."
+        )
+        for operation in migration.operations:
+            command.line(f"  - {operation.describe()}")
+        command.line(f"Migration written to {path}")
+        command.line("Run `nodestream migrations run` to apply the migration.")
diff --git a/nodestream/schema/migrations/migrations.py b/nodestream/schema/migrations/migrations.py
index 2f6874143..c5f4604f1 100644
--- a/nodestream/schema/migrations/migrations.py
+++ b/nodestream/schema/migrations/migrations.py
@@ -179,11 +179,9 @@ def get_ordered_migration_plan(
             # the squashed migration to the plan but will want to add the
             # migrations that were replaced to the plan.
             if (replacement := replacement_index.get(migration.name)) is not None:
-                print("was replaced", migration.name)
                 if not any(
                     r in completed_migration_names for r in replacement.replaces
                 ):
-                    print("skipping", migration.name)
                     continue
 
             # Similarly, if we are looking at a squashed migration, we want to
@@ -192,7 +190,6 @@ def get_ordered_migration_plan(
             if migration.is_squashed_migration():
                 print("squashed", migration.name)
                 if any(r in completed_migration_names for r in migration.replaces):
-                    print("skipping", migration.name)
                     continue
 
             # Now here we are in one of three stats:
@@ -231,6 +228,25 @@ def _iterative_dfs_traversal(self, *start_node: Migration) -> List[Migration]:
 
         return visited_order
 
+    def squash_between(
+        self, name: str, from_migration: Migration, to_migration: Migration
+    ):
+        """Squash all migrations between two migrations.
+
+        Args:
+            name: The name of the new squashed migration.
+            from_migration: The migration to start squashing from.
+            to_migration: The migration to stop squashing at.
+
+        Returns:
+            The new squashed migration.
+        """
+        ordered = self.topological_order()
+        from_index = ordered.index(from_migration)
+        to_index = ordered.index(to_migration)
+        migrations_to_squash = ordered[from_index : to_index + 1]
+        return Migration.squash(name, migrations_to_squash)
+
     def topological_order(self):
         return self._iterative_dfs_traversal(*self.get_leaf_migrations())
 
diff --git a/nodestream/schema/migrations/project_migrations.py b/nodestream/schema/migrations/project_migrations.py
index 59fffe3c5..b6cb5fb38 100644
--- a/nodestream/schema/migrations/project_migrations.py
+++ b/nodestream/schema/migrations/project_migrations.py
@@ -112,3 +112,20 @@ async def create_migration_from_changes(
             return None
         path = migration.write_to_file_with_default_name(self.source_directory)
         return migration, path
+
+    def create_squash_between(
+        self, from_migration: Migration, to_migration: Migration
+    ) -> Tuple[Migration, Path]:
+        """Create a squashed migration between two migrations.
+
+        Args:
+            from_migration: The migration to squash from.
+            to_migration: The migration to squash to.
+
+        Returns:
+            The squashed migration and the path to the file.
+        """
+        name = "squash_from_{}_to_{}".format(from_migration.name, to_migration.name)
+        squashed = self.graph.squash_between(name, from_migration, to_migration)
+        path = squashed.write_to_file_with_default_name(self.source_directory)
+        return squashed, path
diff --git a/tests/unit/cli/commands/test_squash_migrations.py b/tests/unit/cli/commands/test_squash_migrations.py
new file mode 100644
index 000000000..bb146c639
--- /dev/null
+++ b/tests/unit/cli/commands/test_squash_migrations.py
@@ -0,0 +1,17 @@
+import pytest
+from hamcrest import assert_that, equal_to
+
+from nodestream.cli.commands import SquashMigration
+
+
+@pytest.mark.asyncio
+async def test_show_handle_async(mocker):
+    squash = SquashMigration()
+    squash.run_operation = mocker.AsyncMock()
+    squash.get_migrations = mocker.Mock()
+    squash.option = mocker.Mock(
+        side_effect=["from_migration_name", "to_migration_name"]
+    )
+    await squash.handle_async()
+
+    assert_that(squash.run_operation.await_count, equal_to(1))
diff --git a/tests/unit/cli/operations/test_generate_squashed_migration.py b/tests/unit/cli/operations/test_generate_squashed_migration.py
new file mode 100644
index 000000000..f992f4a6c
--- /dev/null
+++ b/tests/unit/cli/operations/test_generate_squashed_migration.py
@@ -0,0 +1,22 @@
+import pytest
+from hamcrest import assert_that, equal_to
+
+from nodestream.cli.commands import NodestreamCommand
+from nodestream.cli.operations import GenerateSquashedMigration
+from nodestream.schema.migrations import Migration
+from nodestream.schema.migrations.operations import CreateNodeType
+
+
+@pytest.mark.asyncio
+async def test_generate_squash_migration_perform(mocker, project_dir):
+    command = mocker.MagicMock(NodestreamCommand)
+    migrations = mocker.Mock()
+    op = CreateNodeType("Person", ["ssn"], [])
+    generated_migration_and_path = Migration("test", [op], []), project_dir
+    migrations.create_squash_between = mocker.Mock(
+        return_value=generated_migration_and_path
+    )
+    subject = GenerateSquashedMigration(migrations, "from", "to")
+    await subject.perform(command)
+
+    assert_that(command.line.call_count, equal_to(5))

From b1505862928ac9a6a413aa900339bc439594a429 Mon Sep 17 00:00:00 2001
From: Zach Probst <Zach_Probst@intuit.com>
Date: Thu, 1 Aug 2024 15:56:35 -0700
Subject: [PATCH 03/11] Get coveraage up

---
 nodestream/schema/migrations/migrations.py          | 13 +------------
 .../schema/migrations/test_project_migrations.py    |  6 ++++++
 2 files changed, 7 insertions(+), 12 deletions(-)

diff --git a/nodestream/schema/migrations/migrations.py b/nodestream/schema/migrations/migrations.py
index c5f4604f1..b582569ae 100644
--- a/nodestream/schema/migrations/migrations.py
+++ b/nodestream/schema/migrations/migrations.py
@@ -86,17 +86,6 @@ def is_squashed_migration(self) -> bool:
         """
         return len(self.replaces) > 0
 
-    def replaces_migration(self, migration: "Migration") -> bool:
-        """Check if this migration replaces another migration.
-
-        Args:
-            migration: The migration to check if this migration replaces.
-
-        Returns:
-            True if this migration replaces the other migration, False otherwise.
-        """
-        return migration.name in self.replaces
-
     @classmethod
     def squash(
         cls,
@@ -127,7 +116,7 @@ def squash(
             }
         )
         if optimize_operations:
-            effective_operations = Operation.optimize_operations(effective_operations)
+            effective_operations = Operation.optimize(effective_operations)
 
         return cls(
             name=new_name,
diff --git a/tests/unit/schema/migrations/test_project_migrations.py b/tests/unit/schema/migrations/test_project_migrations.py
index fce2c7a4d..38f08471e 100644
--- a/tests/unit/schema/migrations/test_project_migrations.py
+++ b/tests/unit/schema/migrations/test_project_migrations.py
@@ -49,3 +49,9 @@ async def test_detect_changes(subject, basic_schema):
     # mocking out a bunch of stuff.
     result = await subject.detect_changes(MigratorInput(), basic_schema)
     assert_that(result.operations, has_length(4))
+
+
+def test_squash_between(subject, root_migration, leaf_migration):
+    migration, path = subject.create_squash_between(root_migration, leaf_migration)
+    assert_that(migration.replaces, has_length(2))
+    assert_that(path.parent, equal_to(subject.source_directory))

From fd9e1cb783eb494368a145b9cd2cf6d60c326d48 Mon Sep 17 00:00:00 2001
From: Zach Probst <Zach_Probst@intuit.com>
Date: Thu, 1 Aug 2024 15:57:49 -0700
Subject: [PATCH 04/11] removing a print

---
 nodestream/schema/migrations/migrations.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/nodestream/schema/migrations/migrations.py b/nodestream/schema/migrations/migrations.py
index b582569ae..c167b014a 100644
--- a/nodestream/schema/migrations/migrations.py
+++ b/nodestream/schema/migrations/migrations.py
@@ -177,7 +177,6 @@ def get_ordered_migration_plan(
             # add it to the plan only if none of the migrations that it replaces
             # have been completed.
             if migration.is_squashed_migration():
-                print("squashed", migration.name)
                 if any(r in completed_migration_names for r in migration.replaces):
                     continue
 

From 156b20ee2674242f89f0d4b87c313a935e321043 Mon Sep 17 00:00:00 2001
From: Zach Probst <Zach_Probst@intuit.com>
Date: Thu, 1 Aug 2024 16:02:22 -0700
Subject: [PATCH 05/11] Remove a literal duplicate

---
 tests/unit/schema/migrations/test_migrations.py | 15 ---------------
 1 file changed, 15 deletions(-)

diff --git a/tests/unit/schema/migrations/test_migrations.py b/tests/unit/schema/migrations/test_migrations.py
index 30f52ceb5..7d7b017f3 100644
--- a/tests/unit/schema/migrations/test_migrations.py
+++ b/tests/unit/schema/migrations/test_migrations.py
@@ -157,21 +157,6 @@ def test_migration_graph_navigates_around_squashed_migrations_when_nothing_appli
     assert_that(plan, is_(equal_to([a, squash, d])))
 
 
-def test_migration_graph_navigates_around_squashed_migrations_when_nothing_applied():
-    a = Migration(name="a", operations=[], dependencies=[])
-    b = Migration(name="b", operations=[], dependencies=["a"])
-    c = Migration(name="c", operations=[], dependencies=["b"])
-    squash = Migration(
-        name="squash", operations=[], dependencies=["a"], replaces=["b", "c"]
-    )
-    d = Migration(name="d", operations=[], dependencies=["squash"])
-
-    graph = MigrationGraph.from_iterable([a, b, c, squash, d])
-
-    plan = graph.get_ordered_migration_plan([])
-    assert_that(plan, is_(equal_to([a, squash, d])))
-
-
 def test_migraiton_graph_handles_dependencies_fulfilled_through_squash():
     a = Migration(name="a", operations=[], dependencies=[])
     b = Migration(name="b", operations=[], dependencies=["a"])

From 28fadc62d4eaf8d1c6687d3363bec763ae390110 Mon Sep 17 00:00:00 2001
From: Zach Probst <Zach_Probst@intuit.com>
Date: Thu, 1 Aug 2024 16:03:11 -0700
Subject: [PATCH 06/11] Placate linter

---
 tests/unit/schema/migrations/test_operations.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/tests/unit/schema/migrations/test_operations.py b/tests/unit/schema/migrations/test_operations.py
index e4996019e..9e7a26d92 100644
--- a/tests/unit/schema/migrations/test_operations.py
+++ b/tests/unit/schema/migrations/test_operations.py
@@ -604,11 +604,11 @@ def test_operation_chain_optimization_complex():
     i = DropRelationshipType("Likes")
     j = CreateNodeType("Team", {"name"}, {})
     k = DropNodeProperty("Team", "mascot")
-    l = DropRelationshipProperty("Likes", "since")
+    dl = DropRelationshipProperty("Likes", "since")
     m = DropNodeType("Team")
 
     # Intertwined and shuffled operations
-    operations = [a, f, b, g, c, j, h, d, k, i, e, l, m]
+    operations = [a, f, b, g, c, j, h, d, k, i, e, dl, m]
 
     assert_that(
         Operation.optimize(operations),

From 541cf4fbb170bb5a64795bf3271f3b4ace25a96e Mon Sep 17 00:00:00 2001
From: Zach Probst <Zach_Probst@intuit.com>
Date: Thu, 1 Aug 2024 16:09:37 -0700
Subject: [PATCH 07/11] Make --to optional and make the cli work

---
 nodestream/cli/commands/squash_migrations.py             | 7 ++-----
 nodestream/cli/operations/generate_squashed_migration.py | 6 +++++-
 nodestream/schema/migrations/migrations.py               | 9 ++++++---
 nodestream/schema/migrations/project_migrations.py       | 7 +++++--
 4 files changed, 18 insertions(+), 11 deletions(-)

diff --git a/nodestream/cli/commands/squash_migrations.py b/nodestream/cli/commands/squash_migrations.py
index e468c2fc8..a896dc2b7 100644
--- a/nodestream/cli/commands/squash_migrations.py
+++ b/nodestream/cli/commands/squash_migrations.py
@@ -14,12 +14,9 @@ class SquashMigration(NodestreamCommand):
             "from",
             description="The name of the migration to squash from.",
             value_required=True,
+            flag=False,
         ),
-        option(
-            "to",
-            description="The name of the migration to squash to.",
-            value_required=True,
-        ),
+        option("to", description="The name of the migration to squash to.", flag=False),
     ]
 
     async def handle_async(self):
diff --git a/nodestream/cli/operations/generate_squashed_migration.py b/nodestream/cli/operations/generate_squashed_migration.py
index 73dd810db..7e71be2ec 100644
--- a/nodestream/cli/operations/generate_squashed_migration.py
+++ b/nodestream/cli/operations/generate_squashed_migration.py
@@ -15,7 +15,11 @@ def __init__(
 
     async def perform(self, command: NodestreamCommand):
         from_migration = self.migrations.graph.get_migration(self.from_migration_name)
-        to_migration = self.migrations.graph.get_migration(self.to_migration_name)
+        to_migration = (
+            self.migrations.graph.get_migration(self.to_migration_name)
+            if self.to_migration_name
+            else None
+        )
         migration, path = self.migrations.create_squash_between(
             from_migration, to_migration
         )
diff --git a/nodestream/schema/migrations/migrations.py b/nodestream/schema/migrations/migrations.py
index c167b014a..9351d5065 100644
--- a/nodestream/schema/migrations/migrations.py
+++ b/nodestream/schema/migrations/migrations.py
@@ -1,6 +1,6 @@
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Dict, Iterable, List
+from typing import Dict, Iterable, List, Optional
 
 from ...file_io import LoadsFromYamlFile, SavesToYamlFile
 from .operations import Operation
@@ -217,7 +217,10 @@ def _iterative_dfs_traversal(self, *start_node: Migration) -> List[Migration]:
         return visited_order
 
     def squash_between(
-        self, name: str, from_migration: Migration, to_migration: Migration
+        self,
+        name: str,
+        from_migration: Migration,
+        to_migration: Optional[Migration] = None,
     ):
         """Squash all migrations between two migrations.
 
@@ -231,7 +234,7 @@ def squash_between(
         """
         ordered = self.topological_order()
         from_index = ordered.index(from_migration)
-        to_index = ordered.index(to_migration)
+        to_index = ordered.index(to_migration) if to_migration else len(ordered) - 1
         migrations_to_squash = ordered[from_index : to_index + 1]
         return Migration.squash(name, migrations_to_squash)
 
diff --git a/nodestream/schema/migrations/project_migrations.py b/nodestream/schema/migrations/project_migrations.py
index b6cb5fb38..69fcff9e6 100644
--- a/nodestream/schema/migrations/project_migrations.py
+++ b/nodestream/schema/migrations/project_migrations.py
@@ -114,7 +114,7 @@ async def create_migration_from_changes(
         return migration, path
 
     def create_squash_between(
-        self, from_migration: Migration, to_migration: Migration
+        self, from_migration: Migration, to_migration: Optional[Migration] = None
     ) -> Tuple[Migration, Path]:
         """Create a squashed migration between two migrations.
 
@@ -125,7 +125,10 @@ def create_squash_between(
         Returns:
             The squashed migration and the path to the file.
         """
-        name = "squash_from_{}_to_{}".format(from_migration.name, to_migration.name)
+        if to_migration is None:
+            name = "squash_from_{}".format(from_migration.name)
+        else:
+            name = "squash_from_{}_to_{}".format(from_migration.name, to_migration.name)
         squashed = self.graph.squash_between(name, from_migration, to_migration)
         path = squashed.write_to_file_with_default_name(self.source_directory)
         return squashed, path

From 916699b5efa766c2af495595ca9383831c97d576 Mon Sep 17 00:00:00 2001
From: Zach Probst <Zach_Probst@intuit.com>
Date: Thu, 1 Aug 2024 16:10:22 -0700
Subject: [PATCH 08/11] Add a new test

---
 tests/unit/schema/migrations/test_project_migrations.py | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/tests/unit/schema/migrations/test_project_migrations.py b/tests/unit/schema/migrations/test_project_migrations.py
index 38f08471e..543de0c87 100644
--- a/tests/unit/schema/migrations/test_project_migrations.py
+++ b/tests/unit/schema/migrations/test_project_migrations.py
@@ -55,3 +55,9 @@ def test_squash_between(subject, root_migration, leaf_migration):
     migration, path = subject.create_squash_between(root_migration, leaf_migration)
     assert_that(migration.replaces, has_length(2))
     assert_that(path.parent, equal_to(subject.source_directory))
+
+
+def test_squash_between_no_to_migration(subject, root_migration):
+    migration, path = subject.create_squash_between(root_migration)
+    assert_that(migration.replaces, has_length(2))
+    assert_that(path.parent, equal_to(subject.source_directory))

From 3d00539ca93eacce09916aca5b02be0de5b39712 Mon Sep 17 00:00:00 2001
From: Zach Probst <zprobst@resilientvitality.com>
Date: Sat, 3 Aug 2024 17:26:37 -0700
Subject: [PATCH 09/11] Simplify test

---
 .../unit/schema/migrations/test_operations.py | 31 +++++++++----------
 1 file changed, 15 insertions(+), 16 deletions(-)

diff --git a/tests/unit/schema/migrations/test_operations.py b/tests/unit/schema/migrations/test_operations.py
index 9e7a26d92..4027fa1ff 100644
--- a/tests/unit/schema/migrations/test_operations.py
+++ b/tests/unit/schema/migrations/test_operations.py
@@ -593,22 +593,21 @@ def test_operation_chain_optimization():
 
 
 def test_operation_chain_optimization_complex():
-    a = CreateNodeType("Person", {"name"}, {"age"})
-    b = CreateNodeType("Sport", {"name"}, {})
-    c = AddNodeProperty("Person", "full_name")
-    d = DropNodeProperty("Person", "age")
-    e = DropNodeType("Person")
-    f = CreateRelationshipType("Likes", {}, {})
-    g = AddRelationshipProperty("Likes", "since")
-    h = AddNodeProperty("Team", "mascot")
-    i = DropRelationshipType("Likes")
-    j = CreateNodeType("Team", {"name"}, {})
-    k = DropNodeProperty("Team", "mascot")
-    dl = DropRelationshipProperty("Likes", "since")
-    m = DropNodeType("Team")
-
-    # Intertwined and shuffled operations
-    operations = [a, f, b, g, c, j, h, d, k, i, e, dl, m]
+    operations = [
+        CreateNodeType("Person", {"name"}, {"age"}),
+        CreateRelationshipType("Likes", {}, {}),
+        CreateNodeType("Sport", {"name"}, {}),
+        AddRelationshipProperty("Likes", "since"),
+        AddNodeProperty("Person", "full_name"),
+        CreateNodeType("Team", {"name"}, {}),
+        DropNodeProperty("Person", "age"),
+        DropRelationshipType("Likes"),
+        DropNodeProperty("Team", "mascot"),
+        DropNodeType("Person"),
+        DropRelationshipProperty("Likes", "since"),
+        AddNodeProperty("Team", "mascot"),
+        DropNodeType("Team"),
+    ]
 
     assert_that(
         Operation.optimize(operations),

From 295dc30a2594afaa54335ddfec88c939710ea033 Mon Sep 17 00:00:00 2001
From: Zach Probst <zprobst@resilientvitality.com>
Date: Sat, 3 Aug 2024 17:31:15 -0700
Subject: [PATCH 10/11] Add a test for migration level

---
 .../unit/schema/migrations/test_migrations.py | 47 ++++++++++++++++++-
 1 file changed, 45 insertions(+), 2 deletions(-)

diff --git a/tests/unit/schema/migrations/test_migrations.py b/tests/unit/schema/migrations/test_migrations.py
index 7d7b017f3..be3891047 100644
--- a/tests/unit/schema/migrations/test_migrations.py
+++ b/tests/unit/schema/migrations/test_migrations.py
@@ -2,7 +2,11 @@
 from hamcrest import assert_that, equal_to, is_
 
 from nodestream.schema.migrations import Migration, MigrationGraph
-from nodestream.schema.migrations.operations import CreateNodeType
+from nodestream.schema.migrations.operations import (
+    AddNodeProperty,
+    CreateNodeType,
+    DropNodeType,
+)
 
 
 def test_migration_is_root_migration(root_migration):
@@ -157,7 +161,7 @@ def test_migration_graph_navigates_around_squashed_migrations_when_nothing_appli
     assert_that(plan, is_(equal_to([a, squash, d])))
 
 
-def test_migraiton_graph_handles_dependencies_fulfilled_through_squash():
+def test_migration_graph_handles_dependencies_fulfilled_through_squash():
     a = Migration(name="a", operations=[], dependencies=[])
     b = Migration(name="b", operations=[], dependencies=["a"])
     c = Migration(name="c", operations=[], dependencies=["b"])
@@ -170,3 +174,42 @@ def test_migraiton_graph_handles_dependencies_fulfilled_through_squash():
 
     plan = graph.get_ordered_migration_plan([a, squash])
     assert_that(plan, is_(equal_to([d])))
+
+
+def test_migration_squash_optimizes_operations():
+    migrations = [
+        Migration(
+            name="a",
+            operations=[CreateNodeType("Person", {"name"}, {"age"})],
+            dependencies=[],
+        ),
+        Migration(
+            name="b",
+            operations=[CreateNodeType("Team", {"name"}, {"mascot"})],
+            dependencies=["a"],
+        ),
+        Migration(
+            name="c",
+            operations=[AddNodeProperty("Person", "full_name")],
+            dependencies=["a"],
+        ),
+        Migration(
+            name="d",
+            operations=[DropNodeType("Person")],
+            dependencies=["c"],
+        ),
+    ]
+
+    assert_that(
+        Migration.squash("bob", migrations),
+        equal_to(
+            Migration(
+                name="bob",
+                operations=[
+                    CreateNodeType("Team", {"name"}, {"mascot"}),
+                ],
+                dependencies=[],
+                replaces=["a", "d", "b", "c"],
+            )
+        ),
+    )

From 9d77282a1eb12d698f1617ba14a3c038159b7fa7 Mon Sep 17 00:00:00 2001
From: Zach Probst <zprobst@resilientvitality.com>
Date: Sat, 3 Aug 2024 17:36:07 -0700
Subject: [PATCH 11/11] Fix flakiness of test

---
 .../unit/schema/migrations/test_migrations.py  | 18 ++++++------------
 1 file changed, 6 insertions(+), 12 deletions(-)

diff --git a/tests/unit/schema/migrations/test_migrations.py b/tests/unit/schema/migrations/test_migrations.py
index be3891047..5bb413748 100644
--- a/tests/unit/schema/migrations/test_migrations.py
+++ b/tests/unit/schema/migrations/test_migrations.py
@@ -1,5 +1,5 @@
 import pytest
-from hamcrest import assert_that, equal_to, is_
+from hamcrest import assert_that, contains_inanyorder, empty, equal_to, is_
 
 from nodestream.schema.migrations import Migration, MigrationGraph
 from nodestream.schema.migrations.operations import (
@@ -200,16 +200,10 @@ def test_migration_squash_optimizes_operations():
         ),
     ]
 
+    result = Migration.squash("bob", migrations)
     assert_that(
-        Migration.squash("bob", migrations),
-        equal_to(
-            Migration(
-                name="bob",
-                operations=[
-                    CreateNodeType("Team", {"name"}, {"mascot"}),
-                ],
-                dependencies=[],
-                replaces=["a", "d", "b", "c"],
-            )
-        ),
+        result.operations, is_(equal_to([CreateNodeType("Team", {"name"}, {"mascot"})]))
     )
+    assert_that(result.dependencies, empty())
+    assert_that(result.replaces, contains_inanyorder("a", "d", "b", "c"))
+    assert_that(result.name, equal_to("bob"))