code: def _reduce(ds: ray.data.Dataset):
      tb_list = traceback.format_list(traceback.extract_stack())
      _already_in_out_of_band_serialization = False
      for tb in tb_list:
          # TODO(xwjiang): Let's make this less hacky.
          if "serialize_lineage" in tb:
              _already_in_out_of_band_serialization = True
              break
      if (
          not _already_in_out_of_band_serialization
          and ds.has_serializable_lineage()
      ):
          return _deserialize_and_fully_execute_if_needed, (
              ds.serialize_lineage(),
          )
      else:
          return ds.__reduce__()
  
imports: import contextlib
  import traceback
  import ray
label_desc: Use image node_id 2 for calling a global function with example usage: _reduce(ds) and returns: _deserialize_and_fully_execute_if_needed,  ds, ds