code: def run_one_op_task(op):
      """Run one task of a PhysicalOperator."""
      tasks = op.get_active_tasks()
      waitable_to_tasks = {task.get_waitable(): task for task in tasks}
      ready, _ = ray.wait(
          list(waitable_to_tasks.keys()),
          num_returns=1,
          fetch_local=False,
      )
      task = waitable_to_tasks[ready[0]]
      if isinstance(task, DataOpTask):
          task.on_data_ready(None)
      else:
          assert isinstance(task, MetadataOpTask)
          task.on_task_finished()
  
file_path: ray/python/ray/data/tests/util.py
imports: import functools
  import os
  import tempfile
  from contextlib import contextmanager
  import ray
  from ray.data._internal.execution.interfaces.physical_operator import DataOpTask, MetadataOpTask, PhysicalOperator
label_desc: Use image node_id 7 for calling a global function with example usage: run_one_op_task(op) without return types