Fix DataflowOutputCounter calculation for ValueInEmptyWindows - #39487
Conversation
When processing shuffle or streaming data in Dataflow Legacy Runner (e.g., from GroupingShuffleReader or WindowingWindmillReader), KeyedWorkItems are wrapped inside a ValueInEmptyWindows (windows.size() == 0). Previously, DataflowOutputCounter.update() counted these as 1 element. This caused inaccurate element counts because: 1. A KeyedWorkItem can contain multiple elements. 2. Elements may belong to multiple windows and need to be fanned out accordingly. 3. KeyedWorkItems containing only timers were incorrectly incrementing element counters.
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
|
Assigning reviewers: R: @jrmccluskey added as fallback since no labels match configuration 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). |
|
Based on the description of the problem, would it make sense to find a way to factor the code so we can have separate streaming/batch implementation of the output counter? The ways they pass elements is just pretty fundamentally different. |
The difference between the streaming and batch output counters really only matters during the GBK step, specifically regarding what gets wrapped inside PTAL! Thanks! |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #39487 +/- ##
============================================
- Coverage 58.05% 55.81% -2.24%
+ Complexity 13057 2272 -10785
============================================
Files 2518 1115 -1403
Lines 263812 174219 -89593
Branches 10765 1460 -9305
============================================
- Hits 153164 97246 -55918
+ Misses 104899 74518 -30381
+ Partials 5749 2455 -3294
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:
|
scwhittle
left a comment
There was a problem hiding this comment.
LGTM with small comment
Thanks!
When processing shuffle or streaming data in Dataflow Legacy Runner (e.g., from GroupingShuffleReader or WindowingWindmillReader), KeyedWorkItems are wrapped inside a ValueInEmptyWindows (windows.size() == 0).
Previously, DataflowOutputCounter.update() counted these as 1 element. This caused inaccurate element counts because:
Internal bug id: 509652238