def METHOD_NAME(self, dataflow_plan: DataflowPlan) -> ExecutionPlan:
    """Convert the dataflow plan to an execution plan."""
    assert len(dataflow_plan.sink_output_nodes) == 1, "Only 1 sink node in the plan is currently supported."
    return dataflow_plan.sink_output_nodes[0].accept_sink_node_visitor(self)