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.
| } | ||
| inputRecordsFromEvents += numRecords - reportedSoFar; | ||
| recordsPerInput[inputIndex] = numRecords; | ||
| approximateInputRecords.setValue((inputRecordsFromEvents * numInputs) / inputsReportingRecords); |
Comment on lines
+911
to
+912
| long records = outputRecordsCounter.getValue() + outputLargeRecordsCounter.getValue(); | ||
| payloadBuilder.setNumRecord((int) Math.min(records, Integer.MAX_VALUE)); |
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
Comment on lines
+164
to
+166
| private long approximateInputRecords(InputContext inputContext) { | ||
| return inputContext.getCounters().findCounter(TaskCounter.APPROXIMATE_INPUT_RECORDS).getValue(); | ||
| } |
Contributor
There was a problem hiding this comment.
this method can go below the @Test methods
| * in the multiplier. A count the writer could not fit in an int arrives negative; ignore it | ||
| * rather than subtract it from the total. | ||
| */ | ||
| public void updateApproximateInputRecords(int inputIndex, int numRecords) { |
Contributor
There was a problem hiding this comment.
nit: this is kind of shuffle internals, a package-protected scope makes more sense to me
I can seen it used to be public, but as the signature is already touched by this PR, maybe make it happen now
Contributor
There was a problem hiding this comment.
consider adding a unit test case for negative numRecords
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
3 times, most recently
from
September 21, 2026 16:02
8c27407 to
9d712ea
Compare
…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 21, 2026 16:12
9d712ea to
f2121aa
Compare
|
💔 -1 overall
This message was automatically generated. |
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.