Make dataflow plan nodes sortable.

metricflow-semantics/metricflow_semantics/dag/mf_dag.py

@@ -32,7 +32,7 @@ class DisplayedProperty:  # type: ignore
     value: Any  # type: ignore
 
 
-@dataclass(frozen=True)
+@dataclass(frozen=True, order=True)
 class NodeId:
     """Unique identifier for nodes in DAGs."""
 

metricflow/dataflow/dataflow_plan.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import functools
 import logging
 import typing
 from abc import ABC, abstractmethod
@@ -23,7 +24,11 @@
 
 NodeSelfT = TypeVar("NodeSelfT", bound="DataflowPlanNode")
 
+# Make it so that we only have to suppress errors here instead of both at the method and the class.
+ComparisonAnyType = typing.Any  # type: ignore[misc]
 
+
+@functools.total_ordering
 @dataclass(frozen=True, eq=False)
 class DataflowPlanNode(DagNode["DataflowPlanNode"], Visitable, ABC):
     """A node in the graph representation of the dataflow.
@@ -81,6 +86,12 @@ def aggregated_to_elements(self) -> Set[LinkableInstanceSpec]:
         """Indicates that the node has been aggregated to these specs, guaranteeing uniqueness in all combinations."""
         return set()
 
+    def __lt__(self, other: ComparisonAnyType) -> bool:  # noqa: D105
+        if not isinstance(other, DataflowPlanNode):
+            raise NotImplementedError
+
+        return self.node_id < other.node_id
+
 
 class DataflowPlan(MetricFlowDag[DataflowPlanNode]):
     """Describes the flow of metric data as it goes from source nodes to sink nodes in the graph."""