Copilot commented on code in PR #2720:
URL: https://github.com/apache/uniffle/pull/2720#discussion_r2734760433
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -478,28 +481,31 @@ protected ShuffleBlockInfo createShuffleBlock(int
partitionId, WriterBuffer wb)
return createDeferredCompressedBlock(partitionId, wb);
}
- byte[] data = wb.getData();
- final int uncompressLength = data.length;
- byte[] compressed = data;
+ final int uncompressLength = wb.getDataLength();
+ final ByteBuf data = wb.getDataAsByteBuf();
+ ByteBuf compressed = data;
if (codec.isPresent()) {
long start = System.currentTimeMillis();
- compressed = codec.get().compress(data);
+ // todo: support ByteBuf compress directly to avoid copying
+ byte[] compressedByteArr = codec.get().compress(wb.getData());
Review Comment:
Calling `wb.getData()` allocates and copies data into a byte array even
though `data` ByteBuf already exists. Consider reusing the already-fetched data
from line 485 by converting the ByteBuf to byte array once, or restructure to
avoid the duplicate call.
```suggestion
byte[] rawData;
if (data.hasArray()
&& data.readerIndex() == 0
&& data.readableBytes() == data.array().length) {
// Reuse the backing array when the ByteBuf fully wraps it
rawData = data.array();
} else {
// Otherwise, copy the readable bytes once into a new array
rawData = new byte[data.readableBytes()];
data.getBytes(data.readerIndex(), rawData);
}
byte[] compressedByteArr = codec.get().compress(rawData);
```
##########
common/src/test/java/org/apache/uniffle/common/util/ChecksumUtilsTest.java:
##########
@@ -112,4 +115,58 @@ public void crc32ByteBufferTest() throws Exception {
directOffsetBuffer.put(data);
assertEquals(expectCrc, ChecksumUtils.getCrc32(directOffsetBuffer, offset,
length));
}
+
+ @Test
+ public void crc32ByteBufEmptyReadableBytesShouldReturnZero() {
+ ByteBuf byteBuf = Unpooled.buffer(16);
+ assertEquals(0, byteBuf.readableBytes());
+ assertEquals(0L, ChecksumUtils.getCrc32(byteBuf));
+ }
+
+ @Test
+ public void crc32ByteBufShouldRespectReaderIndexAndNotChangeIt() {
+ byte[] data = new byte[1024];
+ new Random().nextBytes(data);
+ ByteBuf byteBuf = Unpooled.wrappedBuffer(data);
+
+ int readerIndex = 17;
+ byteBuf.readerIndex(readerIndex);
+
+ CRC32 crc32 = new CRC32();
+ crc32.update(data, readerIndex, data.length - readerIndex);
+ long expected = crc32.getValue();
+
+ assertEquals(expected, ChecksumUtils.getCrc32(byteBuf));
+ assertEquals(readerIndex, byteBuf.readerIndex());
+ }
+
+ @Test
+ public void crc32CompositeByteBufShouldIterateOverNioBuffers() {
+ byte[] part1 = new byte[128];
+ byte[] part2 = new byte[256];
+ Random random = new Random();
+ random.nextBytes(part1);
+ random.nextBytes(part2);
+
+ CompositeByteBuf composite = Unpooled.compositeBuffer();
+ composite.addComponent(true, Unpooled.wrappedBuffer(part1));
+ composite.addComponent(true, Unpooled.wrappedBuffer(part2));
+
+ // Ensure this test hits the composite path (nioBufferCount > 1).
+ // Note: CompositeByteBuf.nioBufferCount() depends on
readerIndex/readableBytes, so check it
+ // here.
+ assertEquals(true, composite.nioBufferCount() > 1);
Review Comment:
Using `assertEquals(true, ...)` is unconventional. Replace with
`assertTrue(composite.nioBufferCount() > 1)` for better readability and clearer
test intent.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
