code: def test_operator_metrics():
      NUM_INPUTS = 100
      NUM_BLOCKS_PER_TASK = 5
      MIN_ROWS_PER_BUNDLE = 10
  
      inputs = make_ref_bundles([[i] for i in range(NUM_INPUTS)])
      input_op = InputDataBuffer(inputs)
  
      def map_fn(block_iter: Iterable[Block], ctx) -> Iterable[Block]:
          for i in range(NUM_BLOCKS_PER_TASK):
              yield pd.DataFrame({"id": [i]})
  
      op = MapOperator.create(
          create_map_transformer_from_block_fn(map_fn),
          input_op=input_op,
          name="TestEstimatedNumBlocks",
          min_rows_per_bundle=MIN_ROWS_PER_BUNDLE,
      )
  
      op.start(ExecutionOptions())
      num_outputs_taken = 0
      bytes_outputs_taken = 0
      for i in range(len(inputs)):
          # Add an input, run all tasks, and take all outputs.
          op.add_input(input_op.get_next(), 0)
          run_op_tasks_sync(op)
          while op.has_next():
              output = op.get_next()
              num_outputs_taken += 1
              bytes_outputs_taken += output.size_bytes()
  
          num_tasks_submitted = (i + 1) // MIN_ROWS_PER_BUNDLE
  
          metrics = op.metrics
          # Check input metrics
          assert metrics.num_inputs_received == i + 1, i
          assert metrics.bytes_inputs_received == sum(
              inputs[k].size_bytes() for k in range(i + 1)
          ), i
          assert (
              metrics.num_inputs_processed
              == num_tasks_submitted * MIN_ROWS_PER_BUNDLE
          ), i
          assert metrics.bytes_inputs_processed == sum(
              inputs[k].size_bytes()
              for k in range(num_tasks_submitted * MIN_ROWS_PER_BUNDLE)
          ), i
  
          # Check outputs metrics
          assert (
              num_outputs_taken
              == num_tasks_submitted * NUM_BLOCKS_PER_TASK
          ), i
          assert metrics.num_outputs_generated == num_outputs_taken, i
          assert (
              metrics.bytes_outputs_generated == bytes_outputs_taken
          ), i
          assert metrics.num_outputs_taken == num_outputs_taken, i
          assert metrics.bytes_outputs_taken == bytes_outputs_taken, i
          assert (
              metrics.num_outputs_of_finished_tasks == num_outputs_taken
          ), i
          assert (
              metrics.bytes_outputs_of_finished_tasks
              == bytes_outputs_taken
          ), i
  
          # Check task metrics
          assert metrics.num_tasks_submitted == num_tasks_submitted, i
          assert metrics.num_tasks_running == 0, i
          assert (
              metrics.num_tasks_have_outputs == num_tasks_submitted
          ), i
          assert metrics.num_tasks_finished == num_tasks_submitted, i
  
          # Check object store metrics
          assert (
              metrics.obj_store_mem_alloc == metrics.bytes_outputs_taken
          ), i
          assert (
              metrics.obj_store_mem_freed
              == metrics.bytes_inputs_processed
          ), i
          assert (
              metrics.obj_store_mem_cur
              == metrics.bytes_inputs_received
              - metrics.bytes_inputs_processed
              + metrics.bytes_outputs_generated
              - metrics.bytes_outputs_taken
          ), i
  
          assert (
              metrics.obj_store_mem_peak >= metrics.obj_store_mem_cur
          ), i
  
imports: import collections
  import random
  import time
  from typing import Any, Iterable, List
  from unittest.mock import MagicMock
  import numpy
  import pandas
  import pytest
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.data._internal.compute import ActorPoolStrategy, TaskPoolStrategy
  from ray.data._internal.execution.interfaces import ExecutionOptions, PhysicalOperator, RefBundle
  from ray.data._internal.execution.operators.actor_pool_map_operator import ActorPoolMapOperator
  from ray.data._internal.execution.operators.base_physical_operator import AllToAllOperator
  from ray.data._internal.execution.operators.input_data_buffer import InputDataBuffer
  from ray.data._internal.execution.operators.limit_operator import LimitOperator
  from ray.data._internal.execution.operators.map_operator import MapOperator, _BlockRefBundler
  from ray.data._internal.execution.operators.map_transformer import create_map_transformer_from_block_fn
  from ray.data._internal.execution.operators.output_splitter import OutputSplitter
  from ray.data._internal.execution.operators.task_pool_map_operator import TaskPoolMapOperator
  from ray.data._internal.execution.operators.union_operator import UnionOperator
  from ray.data._internal.execution.util import make_ref_bundles
  from ray.data.block import Block
  from ray.data.context import DataContext
  from ray.data.tests.util import run_one_op_task, run_op_tasks_sync
  from ray.tests.conftest import *
label_desc: Use image node_id 25 for calling a global function with example usage: test_operator_metrics() without return types