code: def test_op_metrics_logging():
      logger = DatasetLogger(
          "ray.data._internal.execution.streaming_executor"
      ).get_logger()
      with patch.object(logger, "info") as mock_logger:
          ray.data.range(100).map_batches(lambda x: x).materialize()
          logs = [
              canonicalize(call.args[0])
              for call in mock_logger.call_args_list
          ]
          input_str = (
              "Operator InputDataBuffer[Input] completed. Operator Metrics:\n"
              + gen_expected_metrics(is_map=False)
          )
          map_str = (
              "Operator InputDataBuffer[Input] -> "
              "TaskPoolMapOperator[ReadRange->MapBatches(<lambda>)] completed. "
              "Operator Metrics:\n"
          ) + STANDARD_EXTRA_METRICS
  
          # Check that these strings are logged exactly once.
          assert sum([log == input_str for log in logs]) == 1
          assert sum([log == map_str for log in logs]) == 1
  
imports: import re
  import threading
  import time
  from collections import Counter
  from contextlib import contextmanager
  from typing import List, Optional
  from unittest.mock import patch
  import numpy
  import pytest
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.data._internal.dataset_logger import DatasetLogger
  from ray.data._internal.stats import DatasetStats, StatsManager, _get_or_create_stats_actor, _StatsActor
  from ray.data._internal.util import create_dataset_tag
  from ray.data.block import BlockMetadata
  from ray.data.context import DataContext
  from ray.data.tests.util import column_udf
  from ray.tests.conftest import *
label_desc: Use image node_id 30 for calling a global function with example usage: test_op_metrics_logging() without return types