code: def test_map_batches_combine_empty_blocks(ray_start_regular_shared):
      xs = [x % 3 for x in list(range(100))]
  
      # ds1 has 1 block which contains 100 rows.
      ds1 = (
          ray.data.from_items(xs)
          .repartition(1)
          .sort("item")
          .map_batches(lambda x: x)
      )
      assert ds1._block_num_rows() == [100]
  
      # ds2 has 30 blocks, but only 3 of them are non-empty
      ds2 = (
          ray.data.from_items(xs)
          .repartition(30)
          .sort("item")
          .map_batches(lambda x: x, batch_size=1)
      )
      assert len(ds2._block_num_rows()) == 3
      count = sum(1 for x in ds2._block_num_rows() if x > 0)
      assert count == 3
  
      # The number of partitions should not affect the map_batches() result.
      assert ds1.take_all() == ds2.take_all()
  
imports: import itertools
  import math
  import os
  import signal
  import threading
  import time
  from typing import Iterator
  import numpy
  import pandas
  import pyarrow
  import pyarrow.parquet
  import pytest
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.data.block import BlockAccessor
  from ray.data.context import DataContext
  from ray.data.tests.conftest import *
  from ray.data.tests.util import column_udf, column_udf_class, extract_values
  from ray.tests.conftest import *
label_desc: Use image node_id 21 for calling a global function with example usage: test_map_batches_combine_empty_blocks(ray_start_regular_shared) without return types