Cache unbound process fn per review feedback

Author: lakshitbahl

sdks/python/apache_beam/typehints/typecheck.py

@@ -100,15 +100,24 @@ class RuntimeTypeCheckWrapperDoFn(AbstractDoFnWrapper):
   def __init__(self, dofn, type_hints, full_label):
     super().__init__(dofn)
     self.full_label = full_label
-    # Note that the bound process method must not be cached on the instance:
+    # Note that a *bound* process method must not be cached on the instance:
     # an attribute holding a bound method is visible to the stateful DoFn
     # reflection in userstate.get_dofn_specs, and a cached copy can diverge
     # from self.dofn.process across (de)serialization, in which case stateful
-    # DoFn validation would see duplicate StateSpecs/TimerSpecs.
+    # DoFn validation would see duplicate StateSpecs/TimerSpecs. Caching the
+    # underlying (unbound) function is safe: plain functions stored on an
+    # instance are not bound methods, so DoFn reflection ignores them.
+    process_fn = dofn._process_argspec_fn()
+    if hasattr(process_fn, '__func__'):
+      self._process_fn = process_fn.__func__
+      self._process_fn_needs_self = True
+    else:
+      self._process_fn = process_fn
+      self._process_fn_needs_self = False
     if type_hints.input_types:
       input_args, input_kwargs = type_hints.input_types
       self._input_hints = getcallargs_forhints(
-          dofn._process_argspec_fn(), *input_args, **input_kwargs)
+          process_fn, *input_args, **input_kwargs)
     else:
       self._input_hints = None
     # TODO(robertwb): Multi-output.
@@ -125,8 +134,11 @@ def wrapper(self, method, args, kwargs):
   def process(self, *args, **kwargs):
     try:
       if self._input_hints:
-        actual_inputs = inspect.getcallargs(
-            self.dofn._process_argspec_fn(), *args, **kwargs)  # pylint: disable=deprecated-method
+        if self._process_fn_needs_self:
+          actual_inputs = inspect.getcallargs(
+              self._process_fn, self.dofn, *args, **kwargs)  # pylint: disable=deprecated-method
+        else:
+          actual_inputs = inspect.getcallargs(self._process_fn, *args, **kwargs)  # pylint: disable=deprecated-method
         for var, hint in self._input_hints.items():
           if hint is actual_inputs[var]:
             # self parameter