You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by yi...@apache.org on 2022/07/26 09:02:44 UTC
[flink] 02/02: [FLINK-28448] BoundedDataTestBase writes and reads long instead of integer value, makes it easier to compress data for default lz4 compressor
This is an automated email from the ASF dual-hosted git repository.
yingjie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 1b012cb2c402a91be58e8b201ffd1b6d38b58ef9
Author: Weijie Guo <re...@163.com>
AuthorDate: Thu Jul 7 23:18:14 2022 +0800
[FLINK-28448] BoundedDataTestBase writes and reads long instead of integer value, makes it easier to compress data for default lz4 compressor
This closes #20203.
---
.../io/network/buffer/BufferBuilderTestUtils.java | 22 ++++++++++++-
.../io/network/partition/BoundedDataTestBase.java | 38 +++++++++++-----------
2 files changed, 40 insertions(+), 20 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/BufferBuilderTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/BufferBuilderTestUtils.java
index 77eb28d9b68..a8ef32f5224 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/BufferBuilderTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/BufferBuilderTestUtils.java
@@ -27,7 +27,7 @@ import java.nio.ByteOrder;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.junit.Assert.assertEquals;
-/** Utility class for create not-pooled {@link BufferBuilder}. */
+/** Utility class for create {@link BufferBuilder}, {@link BufferConsumer} and {@link Buffer}. */
public class BufferBuilderTestUtils {
public static final int BUFFER_SIZE = 32 * 1024;
@@ -124,6 +124,26 @@ public class BufferBuilderTestUtils {
}
}
+ public static Buffer buildBufferWithAscendingLongs(
+ int bufferSize, int numLongs, long nextValue) {
+ final MemorySegment seg = MemorySegmentFactory.allocateUnpooledSegment(bufferSize);
+ for (int i = 0; i < numLongs; i++) {
+ seg.putLongLittleEndian(8 * i, nextValue++);
+ }
+
+ return new NetworkBuffer(
+ seg, MemorySegment::free, Buffer.DataType.DATA_BUFFER, 8 * numLongs);
+ }
+
+ public static void validateBufferWithAscendingLongs(
+ Buffer buffer, int numLongs, long nextValue) {
+ final ByteBuffer bb = buffer.getNioBufferReadable().order(ByteOrder.LITTLE_ENDIAN);
+
+ for (int i = 0; i < numLongs; i++) {
+ assertEquals(nextValue++, bb.getLong());
+ }
+ }
+
public static Buffer buildSomeBuffer() {
return buildSomeBuffer(1024);
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/BoundedDataTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/BoundedDataTestBase.java
index 401ff02163f..f1849c6993a 100755
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/BoundedDataTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/BoundedDataTestBase.java
@@ -110,11 +110,11 @@ public abstract class BoundedDataTestBase {
}
private void testWriteAndReadData(BoundedData bd) throws Exception {
- final int numInts = 10_000_000;
- final int numBuffers = writeInts(bd, numInts);
+ final int numLongs = 10_000_000;
+ final int numBuffers = writeLongs(bd, numLongs);
bd.finishWrite();
- readInts(bd.createReader(), numBuffers, numInts);
+ readLongs(bd.createReader(), numBuffers, numLongs);
}
@Test
@@ -181,14 +181,14 @@ public abstract class BoundedDataTestBase {
// utils
// ------------------------------------------------------------------------
- private static int writeInts(BoundedData bd, int numInts) throws IOException {
- final int numIntsInBuffer = BUFFER_SIZE / 4;
+ private static int writeLongs(BoundedData bd, int numLongs) throws IOException {
+ final int numLongsInBuffer = BUFFER_SIZE / Long.BYTES;
int numBuffers = 0;
- for (int nextValue = 0; nextValue < numInts; nextValue += numIntsInBuffer) {
+ for (long nextValue = 0; nextValue < numLongs; nextValue += numLongsInBuffer) {
Buffer buffer =
- BufferBuilderTestUtils.buildBufferWithAscendingInts(
- BUFFER_SIZE, numIntsInBuffer, nextValue);
+ BufferBuilderTestUtils.buildBufferWithAscendingLongs(
+ BUFFER_SIZE, numLongsInBuffer, nextValue);
if (compressionEnabled) {
Buffer compressedBuffer = COMPRESSOR.compressToIntermediateBuffer(buffer);
bd.writeBuffer(compressedBuffer);
@@ -206,34 +206,34 @@ public abstract class BoundedDataTestBase {
return numBuffers;
}
- private static void readInts(BoundedData.Reader reader, int numBuffersExpected, int numInts)
+ private static void readLongs(BoundedData.Reader reader, int numBuffersExpected, int numLongs)
throws IOException {
Buffer b;
- int nextValue = 0;
+ long nextValue = 0;
int numBuffers = 0;
- int numIntsInBuffer;
+ int numLongsInBuffer;
while ((b = reader.nextBuffer()) != null) {
if (compressionEnabled && b.isCompressed()) {
Buffer decompressedBuffer = DECOMPRESSOR.decompressToIntermediateBuffer(b);
- numIntsInBuffer = decompressedBuffer.getSize() / 4;
- BufferBuilderTestUtils.validateBufferWithAscendingInts(
- decompressedBuffer, numIntsInBuffer, nextValue);
+ numLongsInBuffer = decompressedBuffer.getSize() / Long.BYTES;
+ BufferBuilderTestUtils.validateBufferWithAscendingLongs(
+ decompressedBuffer, numLongsInBuffer, nextValue);
// recycle intermediate buffer.
decompressedBuffer.recycleBuffer();
} else {
- numIntsInBuffer = b.getSize() / 4;
- BufferBuilderTestUtils.validateBufferWithAscendingInts(
- b, numIntsInBuffer, nextValue);
+ numLongsInBuffer = b.getSize() / Long.BYTES;
+ BufferBuilderTestUtils.validateBufferWithAscendingLongs(
+ b, numLongsInBuffer, nextValue);
}
- nextValue += numIntsInBuffer;
+ nextValue += numLongsInBuffer;
numBuffers++;
b.recycleBuffer();
}
assertEquals(numBuffersExpected, numBuffers);
- assertThat(nextValue, Matchers.greaterThanOrEqualTo(numInts));
+ assertThat(nextValue, Matchers.greaterThanOrEqualTo((long) numLongs));
}
private static Path createTempPath() throws IOException {