diff --git a/metricflow/dataflow/dataflow_plan.py b/metricflow/dataflow/dataflow_plan.py
index dd5f1088b9..1509a29495 100644
--- a/metricflow/dataflow/dataflow_plan.py
+++ b/metricflow/dataflow/dataflow_plan.py
@@ -6,7 +6,7 @@
 import typing
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import FrozenSet, Generic, Optional, Sequence, Set, Type, TypeVar
+from typing import FrozenSet, Optional, Sequence, Set, Type, TypeVar
 
 import more_itertools
 from metricflow_semantics.dag.id_prefix import StaticIdPrefix
@@ -17,27 +17,7 @@
     from dbt_semantic_interfaces.references import SemanticModelReference
     from metricflow_semantics.specs.instance_spec import LinkableInstanceSpec
 
-    from metricflow.dataflow.nodes.add_generated_uuid import AddGeneratedUuidColumnNode
-    from metricflow.dataflow.nodes.aggregate_measures import AggregateMeasuresNode
-    from metricflow.dataflow.nodes.combine_aggregated_outputs import CombineAggregatedOutputsNode
-    from metricflow.dataflow.nodes.compute_metrics import ComputeMetricsNode
-    from metricflow.dataflow.nodes.constrain_time import ConstrainTimeRangeNode
-    from metricflow.dataflow.nodes.filter_elements import FilterElementsNode
-    from metricflow.dataflow.nodes.join_conversion_events import JoinConversionEventsNode
-    from metricflow.dataflow.nodes.join_over_time import JoinOverTimeRangeNode
-    from metricflow.dataflow.nodes.join_to_base import JoinOnEntitiesNode
-    from metricflow.dataflow.nodes.join_to_custom_granularity import JoinToCustomGranularityNode
-    from metricflow.dataflow.nodes.join_to_time_spine import JoinToTimeSpineNode
-    from metricflow.dataflow.nodes.metric_time_transform import MetricTimeDimensionTransformNode
-    from metricflow.dataflow.nodes.min_max import MinMaxNode
-    from metricflow.dataflow.nodes.order_by_limit import OrderByLimitNode
-    from metricflow.dataflow.nodes.read_sql_source import ReadSqlSourceNode
-    from metricflow.dataflow.nodes.semi_additive_join import SemiAdditiveJoinNode
-    from metricflow.dataflow.nodes.where_filter import WhereConstraintNode
-    from metricflow.dataflow.nodes.window_reaggregation_node import WindowReaggregationNode
-    from metricflow.dataflow.nodes.write_to_data_table import WriteToResultDataTableNode
-    from metricflow.dataflow.nodes.write_to_table import WriteToResultTableNode
-
+    from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 
 logger = logging.getLogger(__name__)
 
@@ -102,97 +82,6 @@ def aggregated_to_elements(self) -> Set[LinkableInstanceSpec]:
         return set()
 
 
-class DataflowPlanNodeVisitor(Generic[VisitorOutputT], ABC):
-    """An object that can be used to visit the nodes of a dataflow plan.
-
-    Follows the visitor pattern: https://en.wikipedia.org/wiki/Visitor_pattern
-    All visit* methods are similar and one exists for every type of node in the dataflow plan. The appropriate method
-    will be called with DataflowPlanNode.accept().
-    """
-
-    @abstractmethod
-    def visit_source_node(self, node: ReadSqlSourceNode) -> VisitorOutputT:  # noqa: D102
-        pass
-
-    @abstractmethod
-    def visit_join_on_entities_node(self, node: JoinOnEntitiesNode) -> VisitorOutputT:  # noqa: D102
-        pass
-
-    @abstractmethod
-    def visit_aggregate_measures_node(self, node: AggregateMeasuresNode) -> VisitorOutputT:  # noqa: D102
-        pass
-
-    @abstractmethod
-    def visit_compute_metrics_node(self, node: ComputeMetricsNode) -> VisitorOutputT:  # noqa: D102
-        pass
-
-    @abstractmethod
-    def visit_window_reaggregation_node(self, node: WindowReaggregationNode) -> VisitorOutputT:  # noqa: D102
-        pass
-
-    @abstractmethod
-    def visit_order_by_limit_node(self, node: OrderByLimitNode) -> VisitorOutputT:  # noqa: D102
-        pass
-
-    @abstractmethod
-    def visit_where_constraint_node(self, node: WhereConstraintNode) -> VisitorOutputT:  # noqa: D102
-        pass
-
-    @abstractmethod
-    def visit_write_to_result_data_table_node(self, node: WriteToResultDataTableNode) -> VisitorOutputT:  # noqa: D102
-        pass
-
-    @abstractmethod
-    def visit_write_to_result_table_node(self, node: WriteToResultTableNode) -> VisitorOutputT:  # noqa: D102
-        pass
-
-    @abstractmethod
-    def visit_filter_elements_node(self, node: FilterElementsNode) -> VisitorOutputT:  # noqa: D102
-        pass
-
-    @abstractmethod
-    def visit_combine_aggregated_outputs_node(self, node: CombineAggregatedOutputsNode) -> VisitorOutputT:  # noqa: D102
-        pass
-
-    @abstractmethod
-    def visit_constrain_time_range_node(self, node: ConstrainTimeRangeNode) -> VisitorOutputT:  # noqa: D102
-        pass
-
-    @abstractmethod
-    def visit_join_over_time_range_node(self, node: JoinOverTimeRangeNode) -> VisitorOutputT:  # noqa: D102
-        pass
-
-    @abstractmethod
-    def visit_semi_additive_join_node(self, node: SemiAdditiveJoinNode) -> VisitorOutputT:  # noqa: D102
-        pass
-
-    @abstractmethod
-    def visit_metric_time_dimension_transform_node(  # noqa: D102
-        self, node: MetricTimeDimensionTransformNode
-    ) -> VisitorOutputT:  # noqa: D102
-        pass
-
-    @abstractmethod
-    def visit_join_to_time_spine_node(self, node: JoinToTimeSpineNode) -> VisitorOutputT:  # noqa: D102
-        pass
-
-    @abstractmethod
-    def visit_min_max_node(self, node: MinMaxNode) -> VisitorOutputT:  # noqa: D102
-        pass
-
-    @abstractmethod
-    def visit_add_generated_uuid_column_node(self, node: AddGeneratedUuidColumnNode) -> VisitorOutputT:  # noqa: D102
-        pass
-
-    @abstractmethod
-    def visit_join_conversion_events_node(self, node: JoinConversionEventsNode) -> VisitorOutputT:  # noqa: D102
-        pass
-
-    @abstractmethod
-    def visit_join_to_custom_granularity_node(self, node: JoinToCustomGranularityNode) -> VisitorOutputT:  # noqa: D102
-        pass
-
-
 class DataflowPlan(MetricFlowDag[DataflowPlanNode]):
     """Describes the flow of metric data as it goes from source nodes to sink nodes in the graph."""
 
diff --git a/metricflow/dataflow/dataflow_plan_visitor.py b/metricflow/dataflow/dataflow_plan_visitor.py
new file mode 100644
index 0000000000..61b570cd9b
--- /dev/null
+++ b/metricflow/dataflow/dataflow_plan_visitor.py
@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+import typing
+from abc import ABC, abstractmethod
+from typing import Generic
+
+from metricflow_semantics.visitor import VisitorOutputT
+
+if typing.TYPE_CHECKING:
+    from metricflow.dataflow.nodes.add_generated_uuid import AddGeneratedUuidColumnNode
+    from metricflow.dataflow.nodes.aggregate_measures import AggregateMeasuresNode
+    from metricflow.dataflow.nodes.combine_aggregated_outputs import CombineAggregatedOutputsNode
+    from metricflow.dataflow.nodes.compute_metrics import ComputeMetricsNode
+    from metricflow.dataflow.nodes.constrain_time import ConstrainTimeRangeNode
+    from metricflow.dataflow.nodes.filter_elements import FilterElementsNode
+    from metricflow.dataflow.nodes.join_conversion_events import JoinConversionEventsNode
+    from metricflow.dataflow.nodes.join_over_time import JoinOverTimeRangeNode
+    from metricflow.dataflow.nodes.join_to_base import JoinOnEntitiesNode
+    from metricflow.dataflow.nodes.join_to_custom_granularity import JoinToCustomGranularityNode
+    from metricflow.dataflow.nodes.join_to_time_spine import JoinToTimeSpineNode
+    from metricflow.dataflow.nodes.metric_time_transform import MetricTimeDimensionTransformNode
+    from metricflow.dataflow.nodes.min_max import MinMaxNode
+    from metricflow.dataflow.nodes.order_by_limit import OrderByLimitNode
+    from metricflow.dataflow.nodes.read_sql_source import ReadSqlSourceNode
+    from metricflow.dataflow.nodes.semi_additive_join import SemiAdditiveJoinNode
+    from metricflow.dataflow.nodes.where_filter import WhereConstraintNode
+    from metricflow.dataflow.nodes.window_reaggregation_node import WindowReaggregationNode
+    from metricflow.dataflow.nodes.write_to_data_table import WriteToResultDataTableNode
+    from metricflow.dataflow.nodes.write_to_table import WriteToResultTableNode
+
+
+class DataflowPlanNodeVisitor(Generic[VisitorOutputT], ABC):
+    """An object that can be used to visit the nodes of a dataflow plan.
+
+    Follows the visitor pattern: https://en.wikipedia.org/wiki/Visitor_pattern
+    All visit* methods are similar and one exists for every type of node in the dataflow plan. The appropriate method
+    will be called with DataflowPlanNode.accept().
+    """
+
+    @abstractmethod
+    def visit_source_node(self, node: ReadSqlSourceNode) -> VisitorOutputT:  # noqa: D102
+        pass
+
+    @abstractmethod
+    def visit_join_on_entities_node(self, node: JoinOnEntitiesNode) -> VisitorOutputT:  # noqa: D102
+        pass
+
+    @abstractmethod
+    def visit_aggregate_measures_node(self, node: AggregateMeasuresNode) -> VisitorOutputT:  # noqa: D102
+        pass
+
+    @abstractmethod
+    def visit_compute_metrics_node(self, node: ComputeMetricsNode) -> VisitorOutputT:  # noqa: D102
+        pass
+
+    @abstractmethod
+    def visit_window_reaggregation_node(self, node: WindowReaggregationNode) -> VisitorOutputT:  # noqa: D102
+        pass
+
+    @abstractmethod
+    def visit_order_by_limit_node(self, node: OrderByLimitNode) -> VisitorOutputT:  # noqa: D102
+        pass
+
+    @abstractmethod
+    def visit_where_constraint_node(self, node: WhereConstraintNode) -> VisitorOutputT:  # noqa: D102
+        pass
+
+    @abstractmethod
+    def visit_write_to_result_data_table_node(self, node: WriteToResultDataTableNode) -> VisitorOutputT:  # noqa: D102
+        pass
+
+    @abstractmethod
+    def visit_write_to_result_table_node(self, node: WriteToResultTableNode) -> VisitorOutputT:  # noqa: D102
+        pass
+
+    @abstractmethod
+    def visit_filter_elements_node(self, node: FilterElementsNode) -> VisitorOutputT:  # noqa: D102
+        pass
+
+    @abstractmethod
+    def visit_combine_aggregated_outputs_node(self, node: CombineAggregatedOutputsNode) -> VisitorOutputT:  # noqa: D102
+        pass
+
+    @abstractmethod
+    def visit_constrain_time_range_node(self, node: ConstrainTimeRangeNode) -> VisitorOutputT:  # noqa: D102
+        pass
+
+    @abstractmethod
+    def visit_join_over_time_range_node(self, node: JoinOverTimeRangeNode) -> VisitorOutputT:  # noqa: D102
+        pass
+
+    @abstractmethod
+    def visit_semi_additive_join_node(self, node: SemiAdditiveJoinNode) -> VisitorOutputT:  # noqa: D102
+        pass
+
+    @abstractmethod
+    def visit_metric_time_dimension_transform_node(  # noqa: D102
+        self, node: MetricTimeDimensionTransformNode
+    ) -> VisitorOutputT:  # noqa: D102
+        pass
+
+    @abstractmethod
+    def visit_join_to_time_spine_node(self, node: JoinToTimeSpineNode) -> VisitorOutputT:  # noqa: D102
+        pass
+
+    @abstractmethod
+    def visit_min_max_node(self, node: MinMaxNode) -> VisitorOutputT:  # noqa: D102
+        pass
+
+    @abstractmethod
+    def visit_add_generated_uuid_column_node(self, node: AddGeneratedUuidColumnNode) -> VisitorOutputT:  # noqa: D102
+        pass
+
+    @abstractmethod
+    def visit_join_conversion_events_node(self, node: JoinConversionEventsNode) -> VisitorOutputT:  # noqa: D102
+        pass
+
+    @abstractmethod
+    def visit_join_to_custom_granularity_node(self, node: JoinToCustomGranularityNode) -> VisitorOutputT:  # noqa: D102
+        pass
diff --git a/metricflow/dataflow/nodes/add_generated_uuid.py b/metricflow/dataflow/nodes/add_generated_uuid.py
index 96df59388d..faa2d4f41c 100644
--- a/metricflow/dataflow/nodes/add_generated_uuid.py
+++ b/metricflow/dataflow/nodes/add_generated_uuid.py
@@ -7,7 +7,8 @@
 from metricflow_semantics.dag.mf_dag import DisplayedProperty
 from metricflow_semantics.visitor import VisitorOutputT
 
-from metricflow.dataflow.dataflow_plan import DataflowPlanNode, DataflowPlanNodeVisitor
+from metricflow.dataflow.dataflow_plan import DataflowPlanNode
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 
 
 @dataclass(frozen=True, eq=False)
diff --git a/metricflow/dataflow/nodes/aggregate_measures.py b/metricflow/dataflow/nodes/aggregate_measures.py
index 7fa8c153ce..7162d469e0 100644
--- a/metricflow/dataflow/nodes/aggregate_measures.py
+++ b/metricflow/dataflow/nodes/aggregate_measures.py
@@ -7,7 +7,8 @@
 from metricflow_semantics.specs.measure_spec import MetricInputMeasureSpec
 from metricflow_semantics.visitor import VisitorOutputT
 
-from metricflow.dataflow.dataflow_plan import DataflowPlanNode, DataflowPlanNodeVisitor
+from metricflow.dataflow.dataflow_plan import DataflowPlanNode
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 
 
 @dataclass(frozen=True, eq=False)
diff --git a/metricflow/dataflow/nodes/combine_aggregated_outputs.py b/metricflow/dataflow/nodes/combine_aggregated_outputs.py
index c1c3ad2e3f..68377186b7 100644
--- a/metricflow/dataflow/nodes/combine_aggregated_outputs.py
+++ b/metricflow/dataflow/nodes/combine_aggregated_outputs.py
@@ -8,8 +8,8 @@
 
 from metricflow.dataflow.dataflow_plan import (
     DataflowPlanNode,
-    DataflowPlanNodeVisitor,
 )
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 
 
 @dataclass(frozen=True, eq=False)
diff --git a/metricflow/dataflow/nodes/compute_metrics.py b/metricflow/dataflow/nodes/compute_metrics.py
index 28fb0fd949..6c360b9018 100644
--- a/metricflow/dataflow/nodes/compute_metrics.py
+++ b/metricflow/dataflow/nodes/compute_metrics.py
@@ -12,8 +12,8 @@
 
 from metricflow.dataflow.dataflow_plan import (
     DataflowPlanNode,
-    DataflowPlanNodeVisitor,
 )
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 
 
 @dataclass(frozen=True, eq=False)
diff --git a/metricflow/dataflow/nodes/constrain_time.py b/metricflow/dataflow/nodes/constrain_time.py
index 9694039a44..7247d807f3 100644
--- a/metricflow/dataflow/nodes/constrain_time.py
+++ b/metricflow/dataflow/nodes/constrain_time.py
@@ -8,7 +8,7 @@
 from metricflow_semantics.filters.time_constraint import TimeRangeConstraint
 from metricflow_semantics.visitor import VisitorOutputT
 
-from metricflow.dataflow.dataflow_plan import DataflowPlanNodeVisitor
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 from metricflow.dataflow.nodes.aggregate_measures import DataflowPlanNode
 
 
diff --git a/metricflow/dataflow/nodes/filter_elements.py b/metricflow/dataflow/nodes/filter_elements.py
index 9605f58bdd..abcd5b5bb4 100644
--- a/metricflow/dataflow/nodes/filter_elements.py
+++ b/metricflow/dataflow/nodes/filter_elements.py
@@ -9,7 +9,8 @@
 from metricflow_semantics.specs.spec_set import InstanceSpecSet
 from metricflow_semantics.visitor import VisitorOutputT
 
-from metricflow.dataflow.dataflow_plan import DataflowPlanNode, DataflowPlanNodeVisitor
+from metricflow.dataflow.dataflow_plan import DataflowPlanNode
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 
 
 @dataclass(frozen=True, eq=False)
diff --git a/metricflow/dataflow/nodes/join_conversion_events.py b/metricflow/dataflow/nodes/join_conversion_events.py
index cc6530996b..f17ae1dcc6 100644
--- a/metricflow/dataflow/nodes/join_conversion_events.py
+++ b/metricflow/dataflow/nodes/join_conversion_events.py
@@ -13,7 +13,8 @@
 from metricflow_semantics.specs.time_dimension_spec import TimeDimensionSpec
 from metricflow_semantics.visitor import VisitorOutputT
 
-from metricflow.dataflow.dataflow_plan import DataflowPlanNode, DataflowPlanNodeVisitor
+from metricflow.dataflow.dataflow_plan import DataflowPlanNode
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 
 
 @dataclass(frozen=True, eq=False)
diff --git a/metricflow/dataflow/nodes/join_over_time.py b/metricflow/dataflow/nodes/join_over_time.py
index 92087cf97b..c766bb7dcb 100644
--- a/metricflow/dataflow/nodes/join_over_time.py
+++ b/metricflow/dataflow/nodes/join_over_time.py
@@ -11,7 +11,8 @@
 from metricflow_semantics.specs.time_dimension_spec import TimeDimensionSpec
 from metricflow_semantics.visitor import VisitorOutputT
 
-from metricflow.dataflow.dataflow_plan import DataflowPlanNode, DataflowPlanNodeVisitor
+from metricflow.dataflow.dataflow_plan import DataflowPlanNode
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 
 
 @dataclass(frozen=True, eq=False)
diff --git a/metricflow/dataflow/nodes/join_to_base.py b/metricflow/dataflow/nodes/join_to_base.py
index 50791c580b..bb96ae8c7e 100644
--- a/metricflow/dataflow/nodes/join_to_base.py
+++ b/metricflow/dataflow/nodes/join_to_base.py
@@ -14,7 +14,8 @@
     PartitionDimensionJoinDescription,
     PartitionTimeDimensionJoinDescription,
 )
-from metricflow.dataflow.dataflow_plan import DataflowPlanNode, DataflowPlanNodeVisitor
+from metricflow.dataflow.dataflow_plan import DataflowPlanNode
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 
 
 @dataclass(frozen=True)
diff --git a/metricflow/dataflow/nodes/join_to_custom_granularity.py b/metricflow/dataflow/nodes/join_to_custom_granularity.py
index 2e70b36037..df1ea98efb 100644
--- a/metricflow/dataflow/nodes/join_to_custom_granularity.py
+++ b/metricflow/dataflow/nodes/join_to_custom_granularity.py
@@ -9,7 +9,8 @@
 from metricflow_semantics.specs.time_dimension_spec import TimeDimensionSpec
 from metricflow_semantics.visitor import VisitorOutputT
 
-from metricflow.dataflow.dataflow_plan import DataflowPlanNode, DataflowPlanNodeVisitor
+from metricflow.dataflow.dataflow_plan import DataflowPlanNode
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 
 
 @dataclass(frozen=True, eq=False)
diff --git a/metricflow/dataflow/nodes/join_to_time_spine.py b/metricflow/dataflow/nodes/join_to_time_spine.py
index dfc0f10151..a17b2e4283 100644
--- a/metricflow/dataflow/nodes/join_to_time_spine.py
+++ b/metricflow/dataflow/nodes/join_to_time_spine.py
@@ -14,7 +14,8 @@
 from metricflow_semantics.sql.sql_join_type import SqlJoinType
 from metricflow_semantics.visitor import VisitorOutputT
 
-from metricflow.dataflow.dataflow_plan import DataflowPlanNode, DataflowPlanNodeVisitor
+from metricflow.dataflow.dataflow_plan import DataflowPlanNode
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 
 
 @dataclass(frozen=True, eq=False)
diff --git a/metricflow/dataflow/nodes/metric_time_transform.py b/metricflow/dataflow/nodes/metric_time_transform.py
index 5687904d76..efce6fecf1 100644
--- a/metricflow/dataflow/nodes/metric_time_transform.py
+++ b/metricflow/dataflow/nodes/metric_time_transform.py
@@ -8,7 +8,8 @@
 from metricflow_semantics.dag.mf_dag import DisplayedProperty
 from metricflow_semantics.visitor import VisitorOutputT
 
-from metricflow.dataflow.dataflow_plan import DataflowPlanNode, DataflowPlanNodeVisitor
+from metricflow.dataflow.dataflow_plan import DataflowPlanNode
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 
 
 @dataclass(frozen=True, eq=False)
diff --git a/metricflow/dataflow/nodes/min_max.py b/metricflow/dataflow/nodes/min_max.py
index c7713185f5..597a6466ff 100644
--- a/metricflow/dataflow/nodes/min_max.py
+++ b/metricflow/dataflow/nodes/min_max.py
@@ -6,7 +6,8 @@
 from metricflow_semantics.dag.id_prefix import IdPrefix, StaticIdPrefix
 from metricflow_semantics.visitor import VisitorOutputT
 
-from metricflow.dataflow.dataflow_plan import DataflowPlanNode, DataflowPlanNodeVisitor
+from metricflow.dataflow.dataflow_plan import DataflowPlanNode
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 
 
 @dataclass(frozen=True, eq=False)
diff --git a/metricflow/dataflow/nodes/order_by_limit.py b/metricflow/dataflow/nodes/order_by_limit.py
index f7cbacdf0c..1611536a2e 100644
--- a/metricflow/dataflow/nodes/order_by_limit.py
+++ b/metricflow/dataflow/nodes/order_by_limit.py
@@ -10,8 +10,8 @@
 
 from metricflow.dataflow.dataflow_plan import (
     DataflowPlanNode,
-    DataflowPlanNodeVisitor,
 )
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 
 
 @dataclass(frozen=True, eq=False)
diff --git a/metricflow/dataflow/nodes/read_sql_source.py b/metricflow/dataflow/nodes/read_sql_source.py
index 57a272dffc..c30c6f5578 100644
--- a/metricflow/dataflow/nodes/read_sql_source.py
+++ b/metricflow/dataflow/nodes/read_sql_source.py
@@ -11,7 +11,8 @@
 from metricflow_semantics.visitor import VisitorOutputT
 from typing_extensions import override
 
-from metricflow.dataflow.dataflow_plan import DataflowPlanNode, DataflowPlanNodeVisitor
+from metricflow.dataflow.dataflow_plan import DataflowPlanNode
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 from metricflow.dataset.sql_dataset import SqlDataSet
 
 
diff --git a/metricflow/dataflow/nodes/semi_additive_join.py b/metricflow/dataflow/nodes/semi_additive_join.py
index 1334cde336..46515344ac 100644
--- a/metricflow/dataflow/nodes/semi_additive_join.py
+++ b/metricflow/dataflow/nodes/semi_additive_join.py
@@ -10,7 +10,8 @@
 from metricflow_semantics.specs.time_dimension_spec import TimeDimensionSpec
 from metricflow_semantics.visitor import VisitorOutputT
 
-from metricflow.dataflow.dataflow_plan import DataflowPlanNode, DataflowPlanNodeVisitor
+from metricflow.dataflow.dataflow_plan import DataflowPlanNode
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 
 
 @dataclass(frozen=True, eq=False)
diff --git a/metricflow/dataflow/nodes/where_filter.py b/metricflow/dataflow/nodes/where_filter.py
index 1152376a1d..34785c5445 100644
--- a/metricflow/dataflow/nodes/where_filter.py
+++ b/metricflow/dataflow/nodes/where_filter.py
@@ -8,7 +8,8 @@
 from metricflow_semantics.specs.where_filter.where_filter_spec import WhereFilterSpec
 from metricflow_semantics.visitor import VisitorOutputT
 
-from metricflow.dataflow.dataflow_plan import DataflowPlanNode, DataflowPlanNodeVisitor
+from metricflow.dataflow.dataflow_plan import DataflowPlanNode
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 
 
 @dataclass(frozen=True, eq=False)
diff --git a/metricflow/dataflow/nodes/window_reaggregation_node.py b/metricflow/dataflow/nodes/window_reaggregation_node.py
index 3bbe202c9f..91c064e295 100644
--- a/metricflow/dataflow/nodes/window_reaggregation_node.py
+++ b/metricflow/dataflow/nodes/window_reaggregation_node.py
@@ -12,8 +12,8 @@
 
 from metricflow.dataflow.dataflow_plan import (
     DataflowPlanNode,
-    DataflowPlanNodeVisitor,
 )
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 from metricflow.dataflow.nodes.compute_metrics import ComputeMetricsNode
 
 
diff --git a/metricflow/dataflow/nodes/write_to_data_table.py b/metricflow/dataflow/nodes/write_to_data_table.py
index 66701d7399..c779042580 100644
--- a/metricflow/dataflow/nodes/write_to_data_table.py
+++ b/metricflow/dataflow/nodes/write_to_data_table.py
@@ -8,8 +8,8 @@
 
 from metricflow.dataflow.dataflow_plan import (
     DataflowPlanNode,
-    DataflowPlanNodeVisitor,
 )
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 
 
 @dataclass(frozen=True, eq=False)
diff --git a/metricflow/dataflow/nodes/write_to_table.py b/metricflow/dataflow/nodes/write_to_table.py
index 7a55a5724d..cba37cba8c 100644
--- a/metricflow/dataflow/nodes/write_to_table.py
+++ b/metricflow/dataflow/nodes/write_to_table.py
@@ -9,8 +9,8 @@
 
 from metricflow.dataflow.dataflow_plan import (
     DataflowPlanNode,
-    DataflowPlanNodeVisitor,
 )
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 
 
 @dataclass(frozen=True, eq=False)
diff --git a/metricflow/dataflow/optimizer/predicate_pushdown_optimizer.py b/metricflow/dataflow/optimizer/predicate_pushdown_optimizer.py
index dff6579d0a..ba9e6d7661 100644
--- a/metricflow/dataflow/optimizer/predicate_pushdown_optimizer.py
+++ b/metricflow/dataflow/optimizer/predicate_pushdown_optimizer.py
@@ -15,8 +15,8 @@
 from metricflow.dataflow.dataflow_plan import (
     DataflowPlan,
     DataflowPlanNode,
-    DataflowPlanNodeVisitor,
 )
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 from metricflow.dataflow.nodes.add_generated_uuid import AddGeneratedUuidColumnNode
 from metricflow.dataflow.nodes.aggregate_measures import AggregateMeasuresNode
 from metricflow.dataflow.nodes.combine_aggregated_outputs import CombineAggregatedOutputsNode
diff --git a/metricflow/dataflow/optimizer/source_scan/cm_branch_combiner.py b/metricflow/dataflow/optimizer/source_scan/cm_branch_combiner.py
index 695b349756..e873873855 100644
--- a/metricflow/dataflow/optimizer/source_scan/cm_branch_combiner.py
+++ b/metricflow/dataflow/optimizer/source_scan/cm_branch_combiner.py
@@ -9,8 +9,8 @@
 
 from metricflow.dataflow.dataflow_plan import (
     DataflowPlanNode,
-    DataflowPlanNodeVisitor,
 )
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 from metricflow.dataflow.nodes.add_generated_uuid import AddGeneratedUuidColumnNode
 from metricflow.dataflow.nodes.aggregate_measures import AggregateMeasuresNode
 from metricflow.dataflow.nodes.combine_aggregated_outputs import CombineAggregatedOutputsNode
diff --git a/metricflow/dataflow/optimizer/source_scan/source_scan_optimizer.py b/metricflow/dataflow/optimizer/source_scan/source_scan_optimizer.py
index 9298bb80a2..dd1fe04464 100644
--- a/metricflow/dataflow/optimizer/source_scan/source_scan_optimizer.py
+++ b/metricflow/dataflow/optimizer/source_scan/source_scan_optimizer.py
@@ -11,8 +11,8 @@
 from metricflow.dataflow.dataflow_plan import (
     DataflowPlan,
     DataflowPlanNode,
-    DataflowPlanNodeVisitor,
 )
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 from metricflow.dataflow.nodes.add_generated_uuid import AddGeneratedUuidColumnNode
 from metricflow.dataflow.nodes.aggregate_measures import AggregateMeasuresNode
 from metricflow.dataflow.nodes.combine_aggregated_outputs import CombineAggregatedOutputsNode
diff --git a/metricflow/execution/dataflow_to_execution.py b/metricflow/execution/dataflow_to_execution.py
index 388b0ee1c8..3a438a9f7c 100644
--- a/metricflow/execution/dataflow_to_execution.py
+++ b/metricflow/execution/dataflow_to_execution.py
@@ -8,8 +8,8 @@
 from metricflow.dataflow.dataflow_plan import (
     DataflowPlan,
     DataflowPlanNode,
-    DataflowPlanNodeVisitor,
 )
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 from metricflow.dataflow.nodes.add_generated_uuid import AddGeneratedUuidColumnNode
 from metricflow.dataflow.nodes.aggregate_measures import AggregateMeasuresNode
 from metricflow.dataflow.nodes.combine_aggregated_outputs import CombineAggregatedOutputsNode
diff --git a/metricflow/plan_conversion/dataflow_to_sql.py b/metricflow/plan_conversion/dataflow_to_sql.py
index d283592e84..a723d53bfc 100644
--- a/metricflow/plan_conversion/dataflow_to_sql.py
+++ b/metricflow/plan_conversion/dataflow_to_sql.py
@@ -44,8 +44,8 @@
 
 from metricflow.dataflow.dataflow_plan import (
     DataflowPlanNode,
-    DataflowPlanNodeVisitor,
 )
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 from metricflow.dataflow.nodes.add_generated_uuid import AddGeneratedUuidColumnNode
 from metricflow.dataflow.nodes.aggregate_measures import AggregateMeasuresNode
 from metricflow.dataflow.nodes.combine_aggregated_outputs import CombineAggregatedOutputsNode
diff --git a/tests_metricflow/dataflow/optimizer/source_scan/test_source_scan_optimizer.py b/tests_metricflow/dataflow/optimizer/source_scan/test_source_scan_optimizer.py
index 9af0630d1f..01143c4c18 100644
--- a/tests_metricflow/dataflow/optimizer/source_scan/test_source_scan_optimizer.py
+++ b/tests_metricflow/dataflow/optimizer/source_scan/test_source_scan_optimizer.py
@@ -16,7 +16,8 @@
 from metricflow_semantics.test_helpers.snapshot_helpers import assert_plan_snapshot_text_equal
 
 from metricflow.dataflow.builder.dataflow_plan_builder import DataflowPlanBuilder
-from metricflow.dataflow.dataflow_plan import DataflowPlan, DataflowPlanNode, DataflowPlanNodeVisitor
+from metricflow.dataflow.dataflow_plan import DataflowPlan, DataflowPlanNode
+from metricflow.dataflow.dataflow_plan_visitor import DataflowPlanNodeVisitor
 from metricflow.dataflow.nodes.add_generated_uuid import AddGeneratedUuidColumnNode
 from metricflow.dataflow.nodes.aggregate_measures import AggregateMeasuresNode
 from metricflow.dataflow.nodes.combine_aggregated_outputs import CombineAggregatedOutputsNode