code: def need_reassign_worker(op: OperandType) -> bool:
      # NOTE(qinxuye): special process for reducer
      # We'd better set reducer op's stage to reduce, however,
      # in many case, we copy a reducer op from tileable op,
      # then set stage as reducer one,
      # it would be quite nasty to take over the __setattr__ and
      # make reassign_worker True etc.
      return op.reassign_worker or (
          isinstance(op, MapReduceOperand)
          and op.stage == OperandStage.reduce
      )
  
imports: import itertools
  import logging
  from collections import deque, defaultdict
  from typing import Dict, List, Tuple, Type, Union
  from ....config import Config
  from ....core import ChunkGraph, ChunkType, enter_mode, DAG
  from ....core.operand import Fetch, VirtualOperand, LogicKeyGenerator, MapReduceOperand, OperandStage, ShuffleProxy, ShuffleFetchType
  from ....lib.ordered_set import OrderedSet
  from ....resource import Resource
  from ....typing import BandType, OperandType
  from ....utils import build_fetch, build_fetch_shuffle, tokenize
  from ...subtask import SubtaskGraph, Subtask
  from ..core import Task, new_task_id, MapReduceInfo
  from .assigner import AbstractGraphAssigner, GraphAssigner
  from .fusion import Coloring
label_desc: Use image node_id 1 for calling a global function with example usage: need_reassign_worker(op) and returns: unknown