class_docstr: Tests SamplePartition.
code: def test_sample_metrics(self):
      record_batch = pa.RecordBatch.from_arrays(
          [
              pa.array([["Green"]]),
              pa.array([["Label"]]),
          ],
          ["fa", "label_key"],
      )
      partitioned_rbs = [(1, record_batch)] * 2
      with beam.Pipeline() as p:
          _ = (
              p
              | beam.Create(partitioned_rbs, reshuffle=False)
              | beam.CombinePerKey(
                  partitioned_stats_generator._SampleRecordBatchRows(1)
              )
          )
  
      runner = p.run()
      runner.wait_until_finish()
      actual_metrics = runner.metrics()
      expected_counters = {
          "sample_record_batch_rows_num_instances": 2,
          "sample_record_batch_rows_num_compacts": 1,
      }
      expected_distributions = {
          "sample_record_batch_rows_combine_num_record_batches": {
              "count": 1,
              "sum": 2,
              "max": 2,
              "min": 2,
          },
          "sample_record_batch_rows_combine_byte_size": {
              "count": 1,
              "sum": 42,
              "max": 42,
              "min": 42,
          },
      }
      for counter_name in expected_counters:
          actual_counter = actual_metrics.query(
              beam.metrics.metric.MetricsFilter().with_name(
                  counter_name
              )
          )["counters"]
          self.assertEqual(
              actual_counter[0].committed,
              expected_counters[counter_name],
          )
      for counter_name in expected_distributions:
          actual_counter = actual_metrics.query(
              beam.metrics.metric.MetricsFilter().with_name(
                  counter_name
              )
          )["distributions"]
          self.assertEqual(
              actual_counter[0].committed.count,
              expected_distributions[counter_name]["count"],
          )
          # The size estimation used delegate to Arrow, which empirically
          # sometimes changes how size of batches is calculated.
          self.assertAlmostEqual(
              actual_counter[0].committed.sum,
              expected_distributions[counter_name]["sum"],
              delta=20,
          )
          self.assertAlmostEqual(
              actual_counter[0].committed.max,
              expected_distributions[counter_name]["max"],
              delta=20,
          )
          self.assertAlmostEqual(
              actual_counter[0].committed.min,
              expected_distributions[counter_name]["min"],
              delta=20,
          )
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  from absl.testing import absltest
  from absl.testing import parameterized
  import apache_beam
  from apache_beam.testing import util
  import numpy
  import pyarrow
  from tensorflow_data_validation import constants
  from tensorflow_data_validation import types
  from tensorflow_data_validation.statistics.generators import partitioned_stats_generator
  from tensorflow_data_validation.statistics.generators import sklearn_mutual_information
  from tensorflow_data_validation.utils import test_util
  from google.protobuf import text_format
  from tensorflow.python.util.protobuf import compare
  from tensorflow_metadata.proto.v0 import schema_pb2
  from tensorflow_metadata.proto.v0 import statistics_pb2
label_desc: Use image node_id 3 for calling the SampleRecordBatchRows obj's underlying member method code with example usage: obj.test_sample_metrics() without return types