From 1e71af205b6d0e4c29e17c41199fdeccbfbc4137 Mon Sep 17 00:00:00 2001
From: Danny McCormick <dannymccormick@google.com>
Date: Fri, 6 Nov 2026 20:28:57 +0000
Subject: [PATCH 1/2] Cache __hash__, fast_name, and _TypedMetricName on
 MetricName

---
 sdks/python/apache_beam/metrics/metricbase.py | 39 ++++++++++++++-----
 1 file changed, 29 insertions(+), 10 deletions(-)

diff --git a/sdks/python/apache_beam/metrics/metricbase.py b/sdks/python/apache_beam/metrics/metricbase.py
index addbf14153cf..98863b7d0d03 100644
--- a/sdks/python/apache_beam/metrics/metricbase.py
+++ b/sdks/python/apache_beam/metrics/metricbase.py
@@ -82,9 +82,12 @@ def __init__(
     self.name = name
     self.urn = urn
     self.labels = labels if labels else {}
+    self._cached_fast_name: Optional[str] = None
+    self._cached_hash: Optional[int] = None
+    self._typed_metric_names: dict = {}
 
   def __eq__(self, other):
-    return (
+    return self is other or (
         self.namespace == other.namespace and self.name == other.name and
         self.urn == other.urn and self.labels == other.labels)
 
@@ -97,17 +100,33 @@ def __str__(self):
           self.namespace, self.name)
 
   def __hash__(self):
-    return hash((self.namespace, self.name, self.urn) +
-                tuple(self.labels.items()))
+    if self._cached_hash is None:
+      self._cached_hash = hash(
+          (self.namespace, self.name, self.urn) +
+          tuple(self.labels.items())) if self.labels else hash(
+              (self.namespace, self.name, self.urn))
+    return self._cached_hash
 
   def fast_name(self):
-    name = self.name or ''
-    namespace = self.namespace or ''
-    urn = self.urn or ''
-    labels = ''
-    if self.labels:
-      labels = '_'.join(['%s=%s' % (k, v) for (k, v) in self.labels.items()])
-    return '%d_%s%s%s%s' % (len(name), name, namespace, urn, labels)
+    if self._cached_fast_name is None:
+      name = self.name or ''
+      namespace = self.namespace or ''
+      urn = self.urn or ''
+      labels = ''
+      if self.labels:
+        labels = '_'.join(['%s=%s' % (k, v) for (k, v) in self.labels.items()])
+      self._cached_fast_name = '%d_%s%s%s%s' % (
+          len(name), name, namespace, urn, labels)
+    return self._cached_fast_name
+
+  def __getstate__(self):
+    return (self.namespace, self.name, self.urn, self.labels)
+
+  def __setstate__(self, state):
+    self.namespace, self.name, self.urn, self.labels = state
+    self._cached_fast_name = None
+    self._cached_hash = None
+    self._typed_metric_names = {}
 
 
 class Metric(object):

From 5cd4e050750fd1c34cd8f7908ac4ac5007a9e63d Mon Sep 17 00:00:00 2001
From: Danny McCormick <dannymccormick@google.com>
Date: Fri, 13 Nov 2026 07:01:53 +0000
Subject: [PATCH 2/2] Remove unused _typed_metric_names init and support dict
 state in MetricName.__setstate__

---
 .../python/apache_beam/metrics/metric_test.py | 23 +++++++++++++++++++
 sdks/python/apache_beam/metrics/metricbase.py | 10 +++++---
 2 files changed, 30 insertions(+), 3 deletions(-)

diff --git a/sdks/python/apache_beam/metrics/metric_test.py b/sdks/python/apache_beam/metrics/metric_test.py
index ae66200737b5..90c5a144ca2b 100644
--- a/sdks/python/apache_beam/metrics/metric_test.py
+++ b/sdks/python/apache_beam/metrics/metric_test.py
@@ -57,6 +57,29 @@ def test_basic_metric_name(self):
     self.assertEqual(key.metric.name, 'name1')
     self.assertEqual(key, MetricKey('step1', MetricName('namespace1', 'name1')))
 
+  def test_metric_name_pickle_and_legacy_dict_state(self):
+    import pickle
+    name = MetricName('namespace1', 'name1', urn='urn1', labels={'k': 'v'})
+    _ = hash(name)
+    _ = name.fast_name()
+    unpickled = pickle.loads(pickle.dumps(name))
+    self.assertEqual(unpickled, name)
+    self.assertEqual(hash(unpickled), hash(name))
+    self.assertEqual(unpickled.fast_name(), name.fast_name())
+
+    legacy = MetricName.__new__(MetricName)
+    legacy.__setstate__({
+        'namespace': 'namespace1',
+        'name': 'name1',
+        'urn': 'urn1',
+        'labels': {
+            'k': 'v'
+        },
+    })
+    self.assertEqual(legacy, name)
+    self.assertEqual(hash(legacy), hash(name))
+    self.assertEqual(legacy.fast_name(), name.fast_name())
+
 
 class MetricResultsTest(unittest.TestCase):
   def test_metric_filter_namespace_matching(self):
diff --git a/sdks/python/apache_beam/metrics/metricbase.py b/sdks/python/apache_beam/metrics/metricbase.py
index 98863b7d0d03..d19a1f1c76c3 100644
--- a/sdks/python/apache_beam/metrics/metricbase.py
+++ b/sdks/python/apache_beam/metrics/metricbase.py
@@ -84,7 +84,6 @@ def __init__(
     self.labels = labels if labels else {}
     self._cached_fast_name: Optional[str] = None
     self._cached_hash: Optional[int] = None
-    self._typed_metric_names: dict = {}
 
   def __eq__(self, other):
     return self is other or (
@@ -123,10 +122,15 @@ def __getstate__(self):
     return (self.namespace, self.name, self.urn, self.labels)
 
   def __setstate__(self, state):
-    self.namespace, self.name, self.urn, self.labels = state
+    if isinstance(state, dict):
+      self.namespace = state['namespace']
+      self.name = state['name']
+      self.urn = state['urn']
+      self.labels = state['labels']
+    else:
+      self.namespace, self.name, self.urn, self.labels = state
     self._cached_fast_name = None
     self._cached_hash = None
-    self._typed_metric_names = {}
 
 
 class Metric(object):