Clean up RAY_DATASET_FORCE_LOCAL_METADATA flag (ray-project#32483)

Follow up to ray-project#32015.
cassidylaidlaw · Mar 28, 2023 · 8938d31 · 8938d31
1 parent 1a6d843
commit 8938d31
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 30 deletions.
diff --git a/python/ray/data/dataset.py b/python/ray/data/dataset.py
@@ -1,7 +1,6 @@
 import collections
 import itertools
 import logging
-import os
 import sys
 import time
 import html
@@ -2712,35 +2711,27 @@ def transform(blocks: Iterable[Block], ctx, fn) -> Iterable[Block]:
  datasource.on_write_failed([], e)
  raise
  else:
+ logger.warning(
+ "The Datasource.do_write() is deprecated in "
+ "Ray 2.4 and will be removed in future release. Use "
+ "Datasource.write() instead."
+ )
+
  ctx = DatasetContext.get_current()
  blocks, metadata = zip(*self._plan.execute().get_blocks_with_metadata())
-
- # TODO(ekl) remove this feature flag.
- if "RAY_DATASET_FORCE_LOCAL_METADATA" in os.environ:
- write_results: List[ObjectRef[WriteResult]] = datasource.do_write(
- blocks, metadata, ray_remote_args=ray_remote_args, **write_args
- )
- else:
- logger.warning(
- "The Datasource.do_write() is deprecated in "
- "Ray 2.4 and will be removed in future release. Use "
- "Datasource.write() instead."
- )
- # Prepare write in a remote task so that in Ray client mode, we
- # don't do metadata resolution from the client machine.
- do_write = cached_remote_fn(
- _do_write, retry_exceptions=False, num_cpus=0
- )
- write_results: List[ObjectRef[WriteResult]] = ray.get(
- do_write.remote(
- datasource,
- ctx,
- blocks,
- metadata,
- ray_remote_args,
- _wrap_arrow_serialization_workaround(write_args),
- )
+ # Prepare write in a remote task so that in Ray client mode, we
+ # don't do metadata resolution from the client machine.
+ do_write = cached_remote_fn(_do_write, retry_exceptions=False, num_cpus=0)
+ write_results: List[ObjectRef[WriteResult]] = ray.get(
+ do_write.remote(
+ datasource,
+ ctx,
+ blocks,
+ metadata,
+ ray_remote_args,
+ _wrap_arrow_serialization_workaround(write_args),
  )
+ )
 
  progress = ProgressBar("Write Progress", len(write_results))
  try:

diff --git a/python/ray/data/read_api.py b/python/ray/data/read_api.py
@@ -1,5 +1,4 @@
 import logging
-import os
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, TypeVar, Union
 
 import numpy as np
@@ -279,8 +278,7 @@ def read_datasource(
  ):
  ray_remote_args["scheduling_strategy"] = "SPREAD"
 
- # TODO(ekl) remove this feature flag.
- force_local = "RAY_DATASET_FORCE_LOCAL_METADATA" in os.environ
+ force_local = False
  cur_pg = ray.util.get_current_placement_group()
  pa_ds = _lazy_import_pyarrow_dataset()
  if pa_ds: