code: def _calculate_blocks_rows_and_bytes(
      blocks_with_metadata: BlockPartition,
  ) -> Tuple[List[int], List[int]]:
      """Calculate the number of rows and size in bytes for a list of blocks with
      metadata.
      """
      get_num_rows_and_bytes = cached_remote_fn(_get_num_rows_and_bytes)
      block_rows = []
      block_bytes = []
      for block, metadata in blocks_with_metadata:
          if metadata.num_rows is None or metadata.size_bytes is None:
              # Need to fetch number of rows or size in bytes, so just fetch both.
              num_rows, size_bytes = ray.get(
                  get_num_rows_and_bytes.remote(block)
              )
              # Cache on the block metadata.
              metadata.num_rows = num_rows
              metadata.size_bytes = size_bytes
          block_rows.append(metadata.num_rows)
          block_bytes.append(metadata.size_bytes)
      return block_rows, block_bytes
  
imports: import itertools
  from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
  import ray
  from ray.data._internal.block_list import BlockList
  from ray.data._internal.delegating_block_builder import DelegatingBlockBuilder
  from ray.data._internal.execution.interfaces import TaskContext
  from ray.data._internal.fast_repartition import fast_repartition
  from ray.data._internal.plan import AllToAllStage
  from ray.data._internal.remote_fn import cached_remote_fn
  from ray.data._internal.shuffle_and_partition import PushBasedShufflePartitionOp, SimpleShufflePartitionOp
  from ray.data._internal.sort import SortKey, sort_impl
  from ray.data._internal.split import _split_at_index, _split_at_indices
  from ray.data.block import Block, BlockAccessor, BlockExecStats, BlockMetadata, BlockPartition
  from ray.data.context import DataContext
label_desc: Use image node_id 1 for calling a global function with example usage: _calculate_blocks_rows_and_bytes(blocks_with_metadata) and returns: block_rows,  block_bytes