Copilot commented on code in PR #540:
URL: https://github.com/apache/tez/pull/540#discussion_r4060344229
##########
tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleManager.java:
##########
@@ -348,13 +352,27 @@ public ShuffleManager(InputContext inputContext,
Configuration conf, int numInpu
+ ", asyncHttp=" + asyncHttp);
}
- public void updateApproximateInputRecords(int delta) {
- if (delta <= 0) {
+ /**
+ * Takes the row count one input announces and extrapolates the total over
every input.
+ * An announcement is the source task's output record count so far, not a
per-event delta, so a
+ * pipelined input announces a new total per spill and only its latest one
is kept -- it is the
+ * count at the time the event was built, which the final event brings up to
date. An input that
+ * wrote no rows announces zero and still joins the denominator, which
numInputs already counts
+ * 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) {
+ if (numRecords < 0 || inputIndex < 0 || inputIndex >= numInputs) {
return;
}
- inputRecordsFromEvents += delta;
- eventsReceived++;
- approximateInputRecords.setValue((inputRecordsFromEvents / eventsReceived)
* numInputs);
+ int reportedSoFar = recordsPerInput[inputIndex];
+ if (reportedSoFar == NO_RECORDS_REPORTED) {
+ inputsReportingRecords++;
+ reportedSoFar = 0;
+ }
+ inputRecordsFromEvents += numRecords - reportedSoFar;
+ recordsPerInput[inputIndex] = numRecords;
+ approximateInputRecords.setValue((inputRecordsFromEvents * numInputs) /
inputsReportingRecords);
Review Comment:
This multiplication can overflow `long` before the division. Because each
announced value is capped at `Integer.MAX_VALUE`, a vertex with more than about
65K reporting inputs can produce an incorrect (even negative) counter although
the final extrapolated total still fits in a `long`. Compute the scaled value
with overflow-safe arithmetic (for example, `BigInteger`) instead of
multiplying the two large factors directly.
##########
tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/writers/UnorderedPartitionedKVWriter.java:
##########
@@ -906,7 +906,10 @@ private Event generateDMEvent(boolean addSpillDetails, int
spillId,
DataMovementEventPayloadProto.Builder payloadBuilder =
DataMovementEventPayloadProto
.newBuilder();
if (numPartitions == 1) {
- payloadBuilder.setNumRecord((int) outputRecordsCounter.getValue());
+ // Saturate rather than wrap: the field is an int32 while the counters
are longs. Large
+ // records bypass outputRecordsCounter, so add them the way the
VertexManager event does.
+ long records = outputRecordsCounter.getValue() +
outputLargeRecordsCounter.getValue();
+ payloadBuilder.setNumRecord((int) Math.min(records, Integer.MAX_VALUE));
Review Comment:
The writer tests cover one-partition payloads only for zero/ordinary
records; they do not exercise this new branch with a large record or a count
above `Integer.MAX_VALUE`. Since both the `OUTPUT_LARGE_RECORDS` addition and
saturation are protocol-visible fixes, add a one-partition DME assertion for
those cases so a regression can’t silently reintroduce under-counting or int
wrapping.
--
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]
