Repository navigation
Conversation
|
Assigning reviewers: R: @claudevdm for label python. This pull request likely touches a core component ("core" label). Please review with scrutiny. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #40501 +/- ##
=============================================
+ Coverage 56.16% 79.90% +23.73%
=============================================
Files 1124 488 -636
Lines 178336 82542 -95794
Branches 1489 0 -1489
=============================================
- Hits 100170 65957 -34213
+ Misses 75645 16585 -59060
+ Partials 2521 0 -2521
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
claudevdm
left a comment
There was a problem hiding this comment.
I think this looks fine, but not sure it adresses @damccorm comment from previous #39206 (comment)
| * (YAML) Added WriteToDatadog transform ([#38362](https://github.com/apache/beam/issues/38362)). | ||
| * (Java) Flink 2.1 and 2.2 support is added ([#38947](https://github.com/apache/beam/issues/38947)) ([#38978](https://github.com/apache/beam/issues/38978)); Flink 1.17 and 1.18 support is dropped. | ||
| * (Python) MqttIO is now supported in Python via cross-language ([#21060](https://github.com/apache/beam/issues/21060)). | ||
| * (Python) Reduced runtime type-check overhead by combining input and output checks into a single wrapper ([#20226](https://github.com/apache/beam/issues/20226)). |
There was a problem hiding this comment.
2.75 has been released already. Please move this to future release
|
R: @damccorm do you have any concerns with this? I saw a previous attempt you asked for changes #39206 (comment) |
|
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment |
Reviewing now, I think this one is ok - it is not making other non-functional changes, there are just some changes to make sure we correctly preserve additional variables we're passing into the function Guessing AI caught this distinction both times? But I think it is fine to take it |
6190609 to
72d2196
Compare
|
/gemini review |
There was a problem hiding this comment.
📋 Review Summary
This PR consolidates OutputCheckWrapperDoFn and TypeCheckWrapperDoFn into a single RuntimeTypeCheckWrapperDoFn to eliminate double DoFn wrapper delegation when runtime_type_check is enabled, and avoids caching a bound method on the wrapper instance so userstate.get_dofn_specs does not see duplicate state/timer specs after serialization. There is one bug where non-generator output type violations get wrapped with _add_label twice, duplicating the Runtime type violation detected within ParDo(...): prefix in the error message.
🔍 General Feedback
- The approach of unwrapping
process_fn.__func__so notypes.MethodTypeattribute is stored onRuntimeTypeCheckWrapperDoFn.__dict__cleanly preventsuserstate.get_dofn_specsfrom inspecting_process_fnas a separate DoFn method. - Check the error message formatting on eager (non-generator) output type-check failures, as
_type_check_resultis currently invoked inside atry ... except TypeCheckErrorblock whiletype_check_outputalso wrapsTypeCheckErrorwithself._add_label.
💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.
| # self parameter | ||
| continue | ||
| _check_instance_type(hint, actual_inputs[var], var, True) | ||
|
|
There was a problem hiding this comment.
🟠 High: self._type_check_result(...) is called inside the try ... except TypeCheckError as e: raise self._add_label(e) block in both process() (line 133) and wrapper() (line 114), while type_check_output() inside _type_check_result() (lines 174-177) also catches TypeCheckError and raises self._add_label(e).
When process() or wrapper() returns a non-generator iterable (e.g., a list or tuple, which includes beam.Map and list-returning FlatMap/ParDo transforms), for o in transform_results: type_check_output(o) runs synchronously inside _type_check_result. Any output TypeCheckError is wrapped by self._add_label(e) inside type_check_output and then caught and wrapped by self._add_label(e) a second time in process()/wrapper(), producing a duplicated prefix:
Runtime type violation detected within ParDo(<label>): Runtime type violation detected within ParDo(<label>): ...
Moving self._type_check_result(result) into the else: branch (in both process and wrapper) ensures self._add_label(e) is applied only once for both generator and non-generator outputs.
| result = self.dofn.process(*args, **kwargs) |
There was a problem hiding this comment.
I tested this change with the regression test I just pushed. It fails because the TypeCheckError raised in else bypasses the preceding except, so the ParDo label is lost. I'll keep _type_check_result() inside try.
| result = self._type_check_result(self.dofn.process(*args, **kwargs)) | ||
| except TypeCheckError as e: | ||
| raise self._add_label(e) | ||
| else: |
There was a problem hiding this comment.
🟠 High: Pair with moving self._type_check_result out of the try block on line 133 (and similarly in wrapper on lines 114 and 118) so output TypeCheckErrors are not labeled twice.
| else: | |
| return self._check_type(self._type_check_result(result)) |
There was a problem hiding this comment.
The same issue applies here: exceptions raised in else bypass the preceding except. Since type_check_output() doesn't call _add_label() itself, I couldn't reproduce the double-labeling rationale in my current implementation.
Summary
Fixes #20226
This PR reduces runtime type-checking overhead in the Python SDK by consolidating two nested
DoFnwrappers into a singleRuntimeTypeCheckWrapperDoFn.Previously, input and output checks were handled by separate wrappers. The merged wrapper aims to preserve existing type-checking behavior while eliminating one level of Python call indirection per processed element.
Changes
OutputCheckWrapperDoFnandTypeCheckWrapperDoFnintoRuntimeTypeCheckWrapperDoFn.processfunction while avoiding storing a bound method on the wrapper, preserving statefulDoFnvalidation.DoFncompatibility.Tests
apache_beam/typehints/typecheck_test.py: 18 passed.apache_beam/transforms/ptransform_test.py: 202 passed, 4 skipped.Performance
Compared the original and modified implementations using Beam's existing
runtime_type_check_microbenchmark.Configuration: 20
DoFns and three timing repetitions per configuration.At 2,001 elements, the measured runtime was approximately 4.5% lower for simple type hints and 5.2% lower for nested type hints in this local run.
These measurements are preliminary and may vary by environment. They demonstrate the results of this benchmark configuration, not a guaranteed improvement for all pipelines.