damccorm commented on code in PR #39206:
URL: https://github.com/apache/beam/pull/39206#discussion_r3544592029
##########
sdks/python/apache_beam/typehints/typecheck.py:
##########
@@ -84,25 +84,81 @@ def teardown(self):
return self.dofn.teardown()
-class OutputCheckWrapperDoFn(AbstractDoFnWrapper):
- """A DoFn that verifies against common errors in the output type."""
- def __init__(self, dofn, full_label):
+class RuntimeTypeCheckWrapperDoFn(AbstractDoFnWrapper):
+ """A DoFn wrapper performing runtime type-checking of inputs and outputs.
+
+ This single wrapper performs the work that was previously split between
+ two nested wrappers (``TypeCheckWrapperDoFn`` wrapped inside
+ ``OutputCheckWrapperDoFn``): type-checking of inputs and outputs against
+ declared type hints, verification against common output errors (such as
+ returning a plain ``str``, ``bytes`` or ``dict`` from a ``ParDo``), and
+ labeling of raised ``TypeCheckError`` messages with the offending
+ transform's full label. Merging the wrappers removes one level of Python
+ call indirection per processed element when ``--runtime_type_check`` is
+ enabled.
+ """
+ def __init__(self, dofn, type_hints, full_label):
super().__init__(dofn)
self.full_label = full_label
+ # 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. 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(
+ process_fn, *input_args, **input_kwargs)
+ else:
+ self._input_hints = None
Review Comment:
We've changed the actual logic being used here (this isn't just a refactor).
I think we should avoid mixing refactors and functional changes (if you want to
propose the functional change it should be a separate PR)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
