code: def test_auto_parallelism_placement_group(shutdown_only):
      ray.init(num_cpus=16, num_gpus=8)
  
      @ray.remote
      def run():
          context = DataContext.get_current()
          context.min_parallelism = 1
          ds = ray.data.range_tensor(2000, shape=(100,), parallelism=-1)
          return ds.num_blocks()
  
      # 1/16 * 4 * 16 = 4
      pg = ray.util.placement_group([{"CPU": 1}])
      num_blocks = ray.get(
          run.options(
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=pg
              )
          ).remote()
      )
      assert num_blocks == 4, num_blocks
  
      # 2/16 * 4 * 16 = 8
      pg = ray.util.placement_group([{"CPU": 2}])
      num_blocks = ray.get(
          run.options(
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=pg
              )
          ).remote()
      )
      assert num_blocks == 8, num_blocks
  
      # 1/8 * 4 * 16 = 8
      pg = ray.util.placement_group([{"CPU": 1, "GPU": 1}])
      num_blocks = ray.get(
          run.options(
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=pg
              )
          ).remote()
      )
      assert num_blocks == 8, num_blocks
  
file_path: ray/python/ray/data/tests/test_auto_parallelism.py
imports: from dataclasses import astuple, dataclass
  import pytest
  import ray
  from ray.data._internal.util import _autodetect_parallelism
  from ray.data.context import DataContext
  from ray.tests.conftest import *
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 3 for calling a global function with example usage: test_auto_parallelism_placement_group(shutdown_only) and returns: ds