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:42 UTC
[flink] branch master updated (c7af66be7cb -> 1b012cb2c40)
This is an automated email from the ASF dual-hosted git repository.
yingjie pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from c7af66be7cb [FLINK-28628][table] Introduce Operation Execution Plugin (#20247)
new a4ad7814009 [FLINK-28448] Fix bugs in BoundedDataTestBase when enable compression
new 1b012cb2c40 [FLINK-28448] BoundedDataTestBase writes and reads long instead of integer value, makes it easier to compress data for default lz4 compressor
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../io/network/buffer/BufferBuilderTestUtils.java | 22 ++++++++++-
.../io/network/partition/BoundedDataTestBase.java | 46 +++++++++++++---------
2 files changed, 49 insertions(+), 19 deletions(-)
[flink] 01/02: [FLINK-28448] Fix bugs in BoundedDataTestBase when enable compression
Posted by yi...@apache.org.
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 a4ad7814009f3e799f198d54f24a055894b1ebce
Author: Weijie Guo <re...@163.com>
AuthorDate: Thu Jul 7 23:25:45 2022 +0800
[FLINK-28448] Fix bugs in BoundedDataTestBase when enable compression
1. If enable compression, intermediateBuffer must reduce reference count after write it to file.
2. The logic of using the compressed buffer size to calculate the amount of data is wrong.
This closes #20203.
---
.../runtime/io/network/partition/BoundedDataTestBase.java | 14 ++++++++++++--
1 file changed, 12 insertions(+), 2 deletions(-)
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 8fc1913668f..401ff02163f 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
@@ -190,11 +190,17 @@ public abstract class BoundedDataTestBase {
BufferBuilderTestUtils.buildBufferWithAscendingInts(
BUFFER_SIZE, numIntsInBuffer, nextValue);
if (compressionEnabled) {
- bd.writeBuffer(COMPRESSOR.compressToIntermediateBuffer(buffer));
+ Buffer compressedBuffer = COMPRESSOR.compressToIntermediateBuffer(buffer);
+ bd.writeBuffer(compressedBuffer);
+ // recycle intermediate buffer.
+ if (compressedBuffer != buffer) {
+ compressedBuffer.recycleBuffer();
+ }
} else {
bd.writeBuffer(buffer);
}
numBuffers++;
+ buffer.recycleBuffer();
}
return numBuffers;
@@ -206,13 +212,17 @@ public abstract class BoundedDataTestBase {
int nextValue = 0;
int numBuffers = 0;
+ int numIntsInBuffer;
while ((b = reader.nextBuffer()) != null) {
- final int numIntsInBuffer = b.getSize() / 4;
if (compressionEnabled && b.isCompressed()) {
Buffer decompressedBuffer = DECOMPRESSOR.decompressToIntermediateBuffer(b);
+ numIntsInBuffer = decompressedBuffer.getSize() / 4;
BufferBuilderTestUtils.validateBufferWithAscendingInts(
decompressedBuffer, numIntsInBuffer, nextValue);
+ // recycle intermediate buffer.
+ decompressedBuffer.recycleBuffer();
} else {
+ numIntsInBuffer = b.getSize() / 4;
BufferBuilderTestUtils.validateBufferWithAscendingInts(
b, numIntsInBuffer, nextValue);
}
[flink] 02/02: [FLINK-28448] BoundedDataTestBase writes and reads long instead of integer value, makes it easier to compress data for default lz4 compressor
Posted by yi...@apache.org.
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 {