[data] Remove workaround for ray-project#39151 (ray-project#43757)

Removes the workaround for ray-project#39151. The workaround hides a Ray Core ref counting bug by keeping old refs around. However, this can cause memory leaks.

---------

Signed-off-by: Stephanie Wang <[email protected]>

python/ray/data/_internal/execution/operators/map_operator.py

@@ -77,10 +77,6 @@ def __init__(
         self._metadata_tasks: Dict[int, MetadataOpTask] = {}
         self._next_metadata_task_idx = 0
         # Keep track of all finished streaming generators.
-        # TODO(hchen): This is a workaround for a bug of lineage reconstruction.
-        # When the streaming generator ref is GC'ed, the objects it generated
-        # cannot be reconstructed. Should remove it once Ray Core fixes the bug.
-        self._finished_streaming_gens: List[ObjectRefGenerator] = []
         super().__init__(name, input_op, target_max_block_size)
 
         # If set, then all output blocks will be split into
@@ -316,8 +312,7 @@ def _task_done_callback(task_index: int, exception: Optional[Exception]):
                 / self._metrics.num_tasks_finished
             )
 
-            task = self._data_tasks.pop(task_index)
-            self._finished_streaming_gens.append(task.get_waitable())
+            self._data_tasks.pop(task_index)
             # Notify output queue that this task is complete.
             self._output_queue.notify_task_completed(task_index)
             if task_done_callback:
@@ -385,7 +380,6 @@ def get_map_transformer(self) -> MapTransformer:
     def shutdown(self):
         self._data_tasks.clear()
         self._metadata_tasks.clear()
-        self._finished_streaming_gens.clear()
 
     @abstractmethod
     def current_processor_usage(self) -> ExecutionResources: