abstractdog commented on code in PR #540:
URL: https://github.com/apache/tez/pull/540#discussion_r4143457528
##########
tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/writers/TestUnorderedPartitionedKVWriter.java:
##########
@@ -432,6 +433,72 @@ public void testTextMixedRecordsWithoutFinalMerge(boolean
shouldCompress, Report
textTest(100, 10, 2048, 10, 10, 10, false, false);
}
+ /**
+ * One partition with pipelined shuffle is the only shape where
writeLargeRecord runs while
+ * num_record is set, so it is the only one that can show the DME
under-counting large records.
+ */
+ @ParameterizedTest(name = "test[{0}, {1}]")
+ @MethodSource("data")
+ @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS)
+ public void testLargeRecords_SinglePartition(boolean shouldCompress,
+ ReportPartitionStats reportPartitionStats) throws IOException,
InterruptedException {
+ setupInit(shouldCompress, reportPartitionStats);
+ textTest(0, 1, 2048, 0, 0, 5, true, false);
+ }
+
+ /**
+ * An output small enough to fit one buffer spills only at close, so its
records reach the task
+ * counters in the same call that builds the final event -- which is where a
count read too
+ * early reports zero.
+ */
+ @ParameterizedTest(name = "test[{0}, {1}]")
+ @MethodSource("data")
+ @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS)
+ public void testFinalEventRecordCount_SinglePartition(boolean shouldCompress,
Review Comment:
see https://github.com/apache/tez/pull/540/changes#r4143456360
##########
tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/writers/TestUnorderedPartitionedKVWriter.java:
##########
@@ -432,6 +433,72 @@ public void testTextMixedRecordsWithoutFinalMerge(boolean
shouldCompress, Report
textTest(100, 10, 2048, 10, 10, 10, false, false);
}
+ /**
+ * One partition with pipelined shuffle is the only shape where
writeLargeRecord runs while
+ * num_record is set, so it is the only one that can show the DME
under-counting large records.
+ */
+ @ParameterizedTest(name = "test[{0}, {1}]")
+ @MethodSource("data")
+ @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS)
+ public void testLargeRecords_SinglePartition(boolean shouldCompress,
Review Comment:
I think checkstyle will report this because of the underscore, let's make it
happy by simply writing `testLargeRecordsSinglePartition` (even if I know that
using underscore like this in unit test names is valid)
##########
tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/writers/TestUnorderedPartitionedKVWriter.java:
##########
@@ -432,6 +433,72 @@ public void testTextMixedRecordsWithoutFinalMerge(boolean
shouldCompress, Report
textTest(100, 10, 2048, 10, 10, 10, false, false);
}
+ /**
+ * One partition with pipelined shuffle is the only shape where
writeLargeRecord runs while
+ * num_record is set, so it is the only one that can show the DME
under-counting large records.
+ */
+ @ParameterizedTest(name = "test[{0}, {1}]")
+ @MethodSource("data")
+ @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS)
+ public void testLargeRecords_SinglePartition(boolean shouldCompress,
+ ReportPartitionStats reportPartitionStats) throws IOException,
InterruptedException {
+ setupInit(shouldCompress, reportPartitionStats);
+ textTest(0, 1, 2048, 0, 0, 5, true, false);
+ }
+
+ /**
+ * An output small enough to fit one buffer spills only at close, so its
records reach the task
+ * counters in the same call that builds the final event -- which is where a
count read too
+ * early reports zero.
+ */
+ @ParameterizedTest(name = "test[{0}, {1}]")
+ @MethodSource("data")
+ @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS)
+ public void testFinalEventRecordCount_SinglePartition(boolean shouldCompress,
+ ReportPartitionStats reportPartitionStats) throws IOException,
InterruptedException {
+ setupInit(shouldCompress, reportPartitionStats);
+ textTest(5, 1, 2048, 0, 0, 0, true, false);
+ }
+
+ /**
+ * A single-partition writer is the only one that fills num_record, and
large records never
+ * reach outputRecordsCounter -- writeLargeRecord bypasses it -- so both the
DME and the
+ * VertexManager event have to carry the sum of the two counters. Under
pipelined shuffle the
+ * events are pushed through sendEvents rather than returned by close(), so
both sources are
+ * scanned and the last of each kind is the one that has to be right.
+ */
+ private void assertLastEventsCountEveryRecord(UnorderedPartitionedKVWriter
kvWriter,
+ OutputContext outputContext, List<Event> closeEvents) throws IOException
{
+ List<Event> allEvents = new ArrayList<>(closeEvents);
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<List<Event>> sent = ArgumentCaptor.forClass(List.class);
+ verify(outputContext, atLeast(0)).sendEvents(sent.capture());
+ sent.getAllValues().forEach(allEvents::addAll);
+
+ long expected = kvWriter.outputRecordsCounter.getValue()
+ + kvWriter.outputLargeRecordsCounter.getValue();
Review Comment:
isn't the ground truth somehow derived from `numRecordsWritten`, which is
closer to reality? or assert against both `expected` and `numRecordsWritten`
##########
tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestShuffleInputEventHandlerImpl.java:
##########
@@ -408,6 +408,63 @@ public void testDataMovementEventsWithShuffleData() throws
IOException {
}
}
+ /**
+ * An input whose partition is empty reports zero rows and must still reach
the ShuffleManager:
+ * counting it only in numInputs and not in the denominator is what inflated
the counter.
+ */
+ @Test
+ @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
+ public void testAnEmptyPartitionStillReportsItsRecordCount() throws
IOException {
+ ShuffleManager shuffleManager = mock(ShuffleManager.class);
+ ShuffleInputEventHandlerImpl handler = newHandler(shuffleManager, false);
+
+ handler.handleEvents(Collections.singletonList(
+ createRecordCountEvent(1, 0, createEmptyPartitionByteString(0),
false)));
+
+ verify(shuffleManager).updateApproximateInputRecords(eq(1), eq(0L));
+ verify(shuffleManager).addCompletedInputWithNoData(any());
+ }
+
+ /** The composite path carries the same payload and has to report it too. */
+ @Test
+ @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
+ public void testACompositeEventReportsItsRecordCount() throws IOException {
+ ShuffleManager shuffleManager = mock(ShuffleManager.class);
+ ShuffleInputEventHandlerImpl handler = newHandler(shuffleManager, true);
+
+ handler.handleEvents(Collections.singletonList(createRecordCountEvent(1,
4000, null, true)));
+
+ verify(shuffleManager).updateApproximateInputRecords(eq(1), eq(4000L));
+ }
+
+ /**
+ * A source task can emit more rows than an int32 holds. num_record is an
int64 so the count
+ * survives the wire; passing it as an int would not even compile against
that field.
+ */
+ @Test
+ @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
+ public void testACountAboveIntMaxSurvivesTheWire() throws IOException {
+ ShuffleManager shuffleManager = mock(ShuffleManager.class);
+ ShuffleInputEventHandlerImpl handler = newHandler(shuffleManager, false);
+
+ handler.handleEvents(Collections.singletonList(
+ createRecordCountEvent(1, 3_000_000_000L, null, false)));
+
+ verify(shuffleManager).updateApproximateInputRecords(eq(1),
eq(3_000_000_000L));
+ }
+
+ /** A payload with no record count -- any multi-partition writer -- reports
nothing. */
+ @Test
+ @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
+ public void testAPayloadWithoutARecordCountReportsNothing() throws
IOException {
+ ShuffleManager shuffleManager = mock(ShuffleManager.class);
+ ShuffleInputEventHandlerImpl handler = newHandler(shuffleManager, false);
+
+ handler.handleEvents(Collections.singletonList(createDataMovementEvent(0,
1, null)));
+
+ verify(shuffleManager, times(0)).updateApproximateInputRecords(anyInt(),
anyLong());
+ }
Review Comment:
while I understand the goal of this unit test, nothing strongly shows a
multi-partition writer here, which is confusing, its real goal would be clearer
by having said like something below:
```
/**
* A payload that carries no record count reports nothing: the unset field
is not read as a
* zero, which would count the input as having written no rows.
*/
```
##########
tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/writers/TestUnorderedPartitionedKVWriter.java:
##########
@@ -432,6 +433,72 @@ public void testTextMixedRecordsWithoutFinalMerge(boolean
shouldCompress, Report
textTest(100, 10, 2048, 10, 10, 10, false, false);
}
+ /**
+ * One partition with pipelined shuffle is the only shape where
writeLargeRecord runs while
+ * num_record is set, so it is the only one that can show the DME
under-counting large records.
+ */
+ @ParameterizedTest(name = "test[{0}, {1}]")
+ @MethodSource("data")
+ @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS)
+ public void testLargeRecords_SinglePartition(boolean shouldCompress,
+ ReportPartitionStats reportPartitionStats) throws IOException,
InterruptedException {
+ setupInit(shouldCompress, reportPartitionStats);
+ textTest(0, 1, 2048, 0, 0, 5, true, false);
+ }
+
+ /**
+ * An output small enough to fit one buffer spills only at close, so its
records reach the task
+ * counters in the same call that builds the final event -- which is where a
count read too
+ * early reports zero.
Review Comment:
"which is where a count read too early reports zero", this feels like
mentioning the pre-patch state, could it be improved to focus on what the unit
test actually pins?
##########
tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/writers/TestUnorderedPartitionedKVWriter.java:
##########
@@ -432,6 +433,72 @@ public void testTextMixedRecordsWithoutFinalMerge(boolean
shouldCompress, Report
textTest(100, 10, 2048, 10, 10, 10, false, false);
}
+ /**
+ * One partition with pipelined shuffle is the only shape where
writeLargeRecord runs while
+ * num_record is set, so it is the only one that can show the DME
under-counting large records.
+ */
+ @ParameterizedTest(name = "test[{0}, {1}]")
+ @MethodSource("data")
+ @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS)
+ public void testLargeRecords_SinglePartition(boolean shouldCompress,
+ ReportPartitionStats reportPartitionStats) throws IOException,
InterruptedException {
+ setupInit(shouldCompress, reportPartitionStats);
+ textTest(0, 1, 2048, 0, 0, 5, true, false);
+ }
+
+ /**
+ * An output small enough to fit one buffer spills only at close, so its
records reach the task
+ * counters in the same call that builds the final event -- which is where a
count read too
+ * early reports zero.
+ */
+ @ParameterizedTest(name = "test[{0}, {1}]")
+ @MethodSource("data")
+ @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS)
+ public void testFinalEventRecordCount_SinglePartition(boolean shouldCompress,
+ ReportPartitionStats reportPartitionStats) throws IOException,
InterruptedException {
+ setupInit(shouldCompress, reportPartitionStats);
+ textTest(5, 1, 2048, 0, 0, 0, true, false);
+ }
+
+ /**
+ * A single-partition writer is the only one that fills num_record, and
large records never
+ * reach outputRecordsCounter -- writeLargeRecord bypasses it -- so both the
DME and the
+ * VertexManager event have to carry the sum of the two counters. Under
pipelined shuffle the
+ * events are pushed through sendEvents rather than returned by close(), so
both sources are
+ * scanned and the last of each kind is the one that has to be right.
+ */
+ private void assertLastEventsCountEveryRecord(UnorderedPartitionedKVWriter
kvWriter,
+ OutputContext outputContext, List<Event> closeEvents) throws IOException
{
+ List<Event> allEvents = new ArrayList<>(closeEvents);
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<List<Event>> sent = ArgumentCaptor.forClass(List.class);
+ verify(outputContext, atLeast(0)).sendEvents(sent.capture());
+ sent.getAllValues().forEach(allEvents::addAll);
+
+ long expected = kvWriter.outputRecordsCounter.getValue()
+ + kvWriter.outputLargeRecordsCounter.getValue();
+ long lastDmeCount = -1;
+ long lastVmCount = -1;
+ for (Event event : allEvents) {
+ if (event instanceof CompositeDataMovementEvent) {
+ ShuffleUserPayloads.DataMovementEventPayloadProto dme =
+
ShuffleUserPayloads.DataMovementEventPayloadProto.parseFrom(ByteString.copyFrom(
+ ((CompositeDataMovementEvent) event).getUserPayload()));
+ if (dme.hasNumRecord()) {
+ lastDmeCount = dme.getNumRecord();
+ }
+ } else if (event instanceof VertexManagerEvent) {
+ ShuffleUserPayloads.VertexManagerEventPayloadProto vme =
+
ShuffleUserPayloads.VertexManagerEventPayloadProto.parseFrom(ByteString.copyFrom(
+ ((VertexManagerEvent) event).getUserPayload()));
+ lastVmCount = vme.getNumRecord();
+ }
+ }
+ assertTrue(lastDmeCount >= 0, "expected a DME carrying num_record");
+ assertEquals(expected, lastDmeCount, "DME");
+ assertEquals(expected, lastVmCount, "VertexManagerEvent");
Review Comment:
please improve assertion messages here
##########
tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/TestShuffleUtils.java:
##########
@@ -286,6 +286,36 @@ public void
testGenerateOnSpillEvent_With_All_EmptyPartitions() throws Exception
"emptyPartitionBitSet cardinality (expecting 10) = " +
emptyPartitionsBitSet.cardinality());
}
+ /**
+ * An output that never started delivers no rows, and a single-partition one
fills num_record
+ * like any other, so it joins the consumer's denominator instead of being
scaled over.
+ */
+ @Test
+ public void testNonStartedSinglePartitionOutputReportsZeroRecords() throws
Exception {
+ List<Event> events = Lists.newLinkedList();
+ ShuffleUtils.generateEventsForNonStartedOutput(events, 1, outputContext,
false, true,
+ TezCommonUtils.newBestCompressionDeflater());
+
+ ShuffleUserPayloads.DataMovementEventPayloadProto proto =
Review Comment:
`DataMovementEventPayloadProto` is already imported, I would use it instead
##########
tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/writers/TestUnorderedPartitionedKVWriter.java:
##########
@@ -432,6 +433,72 @@ public void testTextMixedRecordsWithoutFinalMerge(boolean
shouldCompress, Report
textTest(100, 10, 2048, 10, 10, 10, false, false);
}
+ /**
Review Comment:
maybe a missing test case, single partition, no data, pipeline shuffle:
```
textTest(0, 1, 2048, 0, 0, 0, true, false)
```
covering the "all empty partitions" case, look at the `writer.finalSpill`
method
--
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]
