Merge pull request #33195 Simplify metrics in preparation for adding …

…another metric type.

sdks/python/apache_beam/internal/metrics/cells.py

@@ -28,7 +28,6 @@
 from typing import TYPE_CHECKING
 from typing import Optional
 
-from apache_beam.metrics.cells import MetricAggregator
 from apache_beam.metrics.cells import MetricCell
 from apache_beam.metrics.cells import MetricCellFactory
 from apache_beam.utils.histogram import Histogram
@@ -50,10 +49,10 @@ class HistogramCell(MetricCell):
   """
   def __init__(self, bucket_type):
     self._bucket_type = bucket_type
-    self.data = HistogramAggregator(bucket_type).identity_element()
+    self.data = HistogramData.identity_element(bucket_type)
 
   def reset(self):
-    self.data = HistogramAggregator(self._bucket_type).identity_element()
+    self.data = HistogramData.identity_element(self._bucket_type)
 
   def combine(self, other: 'HistogramCell') -> 'HistogramCell':
     result = HistogramCell(self._bucket_type)
@@ -148,22 +147,6 @@ def combine(self, other: Optional['HistogramData']) -> 'HistogramData':
 
     return HistogramData(self.histogram.combine(other.histogram))
 
-
-class HistogramAggregator(MetricAggregator):
-  """For internal use only; no backwards-compatibility guarantees.
-
-  Aggregator for Histogram metric data during pipeline execution.
-
-  Values aggregated should be ``HistogramData`` objects.
-  """
-  def __init__(self, bucket_type: 'BucketType') -> None:
-    self._bucket_type = bucket_type
-
-  def identity_element(self) -> HistogramData:
-    return HistogramData(Histogram(self._bucket_type))
-
-  def combine(self, x: HistogramData, y: HistogramData) -> HistogramData:
-    return x.combine(y)
-
-  def result(self, x: HistogramData) -> HistogramResult:
-    return HistogramResult(x.get_cumulative())
+  @staticmethod
+  def identity_element(bucket_type) -> 'HistogramData':
+    return HistogramData(Histogram(bucket_type))

sdks/python/apache_beam/metrics/cells.pxd

@@ -33,21 +33,26 @@ cdef class CounterCell(MetricCell):
   cpdef bint update(self, value) except -1
 
 
+# Not using AbstractMetricCell so that data can be typed.
 cdef class DistributionCell(MetricCell):
   cdef readonly DistributionData data
 
   @cython.locals(ivalue=libc.stdint.int64_t)
   cdef inline bint _update(self, value) except -1
 
 
-cdef class GaugeCell(MetricCell):
-  cdef readonly object data
+cdef class AbstractMetricCell(MetricCell):
+  cdef readonly object data_class
+  cdef public object data
+  cdef bint _update_locked(self, value) except -1
 
 
-cdef class StringSetCell(MetricCell):
-  cdef readonly object data
+cdef class GaugeCell(AbstractMetricCell):
+  pass
 
-  cdef inline bint _update(self, value) except -1
+
+cdef class StringSetCell(AbstractMetricCell):
+  pass
 
 
 cdef class DistributionData(object):