TEZ-4757: APPROXIMATE_INPUT_RECORDS over-counts the rows an input will deliver - #540
Merged
Merged
Conversation
deniskuzZ
commented
Sep 19, 2026
deniskuzZ
commented
Member
- An announcement carries the source task's output record count so far, not a per-event delta; keep the latest value per input instead of summing events, so a pipelined input's earlier spills are not counted again
- Count an input that wrote no rows in the denominator; leaving it out while numInputs keeps it in the multiplier spread the mean of the inputs that had data over the ones that did not
- Announce from the composite event path too: which path runs depends only on tez.am.shuffle.auxiliary-service.id, so on the Tez shuffle handler the counter was never updated at all
- Multiply before dividing, the mean was truncated first
- Saturate the announced count instead of wrapping it through int, and include large records, which bypass OUTPUT_RECORDS -- the VertexManager event already counts both
- updateApproximateInputRecords takes the input index
- Measured before the fix: 10 inputs, 2 announcing 1000 rows each, read 10,000 instead of 2,000; 4 pipelined events totalling 600 read 450
|
🎊 +1 overall
This message was automatically generated. |
There was a problem hiding this comment.
Copilot review overview
🟡 Changes recommended
One or more issues must be addressed before approval.
Get a fresh assessment by requesting another Copilot review.
Review effort: Lite
Findings: 1
Open (2)
What changed in this PR
Fixes approximate input record estimation for unordered shuffle inputs.
Changes:
- Track the latest per-input record total, including empty inputs.
- Update both regular and composite event paths.
- Include large records and saturate payload counts.
| File | Description |
|---|---|
| tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestShuffleManager.java | Updated as part of this pull request. |
| tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestShuffleInputEventHandlerImpl.java | Updated as part of this pull request. |
| tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/writers/UnorderedPartitionedKVWriter.java | Updated as part of this pull request. |
| tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleManager.java | Updated as part of this pull request. |
| tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleInputEventHandlerImpl.java | Updated as part of this pull request. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
abstractdog
requested changes
Sep 21, 2026
abstractdog
left a comment
abstractdog
left a comment
Contributor
There was a problem hiding this comment.
nice catch, left minor comments only
check also copilot comments please, if they make sense
deniskuzZ
force-pushed
the
TEZ-approx-input-records
branch
3 times, most recently
from
September 21, 2026 13:24
a03e61c to
9271332
Compare
|
💔 -1 overall
This message was automatically generated. |
deniskuzZ
force-pushed
the
TEZ-approx-input-records
branch
4 times, most recently
from
September 21, 2026 16:12
9d712ea to
f2121aa
Compare
|
💔 -1 overall
This message was automatically generated. |
deniskuzZ
force-pushed
the
TEZ-approx-input-records
branch
from
September 22, 2026 20:55
f2121aa to
9d9a96c
Compare
|
💔 -1 overall
This message was automatically generated. |
deniskuzZ
force-pushed
the
TEZ-approx-input-records
branch
from
September 24, 2026 13:58
9d9a96c to
3d4487c
Compare
|
💔 -1 overall
This message was automatically generated. |
…l deliver - A report carries the source task's output record count so far, not a per-event delta; keep the latest value per input instead of summing events, so a pipelined input's earlier spills are not counted again - Ignore a report smaller than that input's last one: spill callbacks read the counter and send their event with no lock between, so a stale total can arrive last and would otherwise subtract from the sum - Count an input that wrote no rows in the denominator; leaving it out while numInputs keeps it in the multiplier spread the mean of the inputs that had data over the ones that did not - An output that never started reports 0 rather than nothing, so it joins the denominator instead of being scaled over - Flush the record counters before the final pipelined events read them. An output small enough for one buffer spills only at close, so its DME and its VertexManager event both carried 0 - Report from the composite event path too: which path runs depends only on tez.am.shuffle.auxiliary-service.id, so on the Tez shuffle handler the counter was never updated at all - Scale the mean without forming sum * numInputs, so only the true answer has to fit in a long - num_record is int64, matching the VertexManagerEventPayloadProto field it mirrors. It was int32 while both are filled from long counters, so a task emitting more than 2^31 rows wrapped negative; widening the field alone would only move the loss to the consumer, so the per-input array and the update method carry a long as well. int32 -> int64 is wire compatible in both directions on the same field number - The count includes OUTPUT_LARGE_RECORDS, which bypass OUTPUT_RECORDS; this is what ShuffleUtils.generateVMEvent already sends. Without it a single-partition writer reports 0 for a broadcast of large rows - updateApproximateInputRecords is package-private: it is shuffle internals with one caller in the same package - Measured before the fix: 10 inputs, 2 reporting 1000 rows each, read 10,000 instead of 2,000; 4 pipelined events totalling 600 read 450; a 5-row single-partition pipelined output reported 0
deniskuzZ
force-pushed
the
TEZ-approx-input-records
branch
from
September 30, 2026 11:00
3d4487c to
24d7eeb
Compare
abstractdog
requested changes
Sep 30, 2026
abstractdog
left a comment
abstractdog
left a comment
Contributor
There was a problem hiding this comment.
patch looks good, minor comments, mostly about testing
deniskuzZ
force-pushed
the
TEZ-approx-input-records
branch
from
September 30, 2026 12:34
24d7eeb to
a401d9a
Compare
deniskuzZ
force-pushed
the
TEZ-approx-input-records
branch
from
September 30, 2026 12:58
a401d9a to
09fa1a9
Compare
- A negative report is rejected: it neither enters the sum nor marks the input as reported - Writer tests assert against the records the test wrote, with messages naming the event and counter that disagree; the no-data single-partition pipelined case is pinned in the harness that already runs it - Underscore-free test names, javadocs state what is pinned, helper below the tests, imported proto type used unqualified
deniskuzZ
force-pushed
the
TEZ-approx-input-records
branch
from
September 30, 2026 13:04
09fa1a9 to
e22cc8a
Compare
|
🎊 +1 overall
This message was automatically generated. |
Member
Author
|
thank for the review @abstractdog! |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.