You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ga...@apache.org on 2024/10/18 13:44:26 UTC
(flink) branch release-1.18 updated: [FLINK-36530][state] Fix S3 performance issue with uncompressed state restore
This is an automated email from the ASF dual-hosted git repository.
gaborgsomogyi pushed a commit to branch release-1.18
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.18 by this push:
new a38396fbcea [FLINK-36530][state] Fix S3 performance issue with uncompressed state restore
a38396fbcea is described below
commit a38396fbceaa88992103b79feff71acd7b83e54b
Author: Gabor Somogyi <ga...@apple.com>
AuthorDate: Fri Oct 18 15:44:19 2024 +0200
[FLINK-36530][state] Fix S3 performance issue with uncompressed state restore
---
.../state/CompressibleFSDataInputStream.java | 8 +++++---
.../state/CompressibleFSDataInputStreamTest.java | 22 ++++++++++++++++------
2 files changed, 21 insertions(+), 9 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataInputStream.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataInputStream.java
index dc95b0dc379..c5885b637d1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataInputStream.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataInputStream.java
@@ -31,19 +31,21 @@ public class CompressibleFSDataInputStream extends FSDataInputStream {
private final FSDataInputStream delegate;
private final InputStream compressingDelegate;
+ private final boolean compressed;
public CompressibleFSDataInputStream(
FSDataInputStream delegate, StreamCompressionDecorator compressionDecorator)
throws IOException {
this.delegate = delegate;
this.compressingDelegate = compressionDecorator.decorateWithCompression(delegate);
+ this.compressed = compressionDecorator != UncompressedStreamCompressionDecorator.INSTANCE;
}
@Override
public void seek(long desired) throws IOException {
- final int available = compressingDelegate.available();
- if (available > 0) {
- if (available != compressingDelegate.skip(available)) {
+ if (compressed) {
+ final int available = compressingDelegate.available();
+ if (available > 0 && available != compressingDelegate.skip(available)) {
throw new IOException("Unable to skip buffered data.");
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/CompressibleFSDataInputStreamTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/CompressibleFSDataInputStreamTest.java
index 085afa69bff..9c32ccf1f67 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/CompressibleFSDataInputStreamTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/CompressibleFSDataInputStreamTest.java
@@ -23,7 +23,9 @@ import org.apache.flink.core.fs.FSDataInputStream;
import org.apache.flink.core.fs.FSDataOutputStream;
import org.apache.flink.core.memory.ByteArrayOutputStreamWithPos;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import java.io.ByteArrayInputStream;
import java.io.IOException;
@@ -33,6 +35,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
@@ -99,8 +102,15 @@ class CompressibleFSDataInputStreamTest {
assertThat(readBuffer).asString(StandardCharsets.UTF_8).isEqualTo(prefix);
}
- @Test
- void testSeek() throws IOException {
+ private static Stream<Arguments> testSeekParameters() {
+ return Stream.of(
+ Arguments.of(new UncompressedStreamCompressionDecorator()),
+ Arguments.of(new SnappyStreamCompressionDecorator()));
+ }
+
+ @ParameterizedTest
+ @MethodSource("testSeekParameters")
+ void testSeek(StreamCompressionDecorator streamCompressionDecorator) throws IOException {
final List<String> records = Arrays.asList("first", "second", "third", "fourth", "fifth");
final Map<String, Long> positions = new HashMap<>();
@@ -108,7 +118,7 @@ class CompressibleFSDataInputStreamTest {
try (final TestingOutputStream outputStream = new TestingOutputStream();
final CompressibleFSDataOutputStream compressibleOutputStream =
new CompressibleFSDataOutputStream(
- outputStream, new SnappyStreamCompressionDecorator())) {
+ outputStream, streamCompressionDecorator)) {
for (String record : records) {
positions.put(record, compressibleOutputStream.getPos());
compressibleOutputStream.write(record.getBytes(StandardCharsets.UTF_8));
@@ -121,7 +131,7 @@ class CompressibleFSDataInputStreamTest {
new InputStreamFSInputWrapper(new ByteArrayInputStream(compressedBytes));
final FSDataInputStream compressibleInputStream =
new CompressibleFSDataInputStream(
- inputStream, new SnappyStreamCompressionDecorator())) {
+ inputStream, streamCompressionDecorator)) {
verifyRecord(compressibleInputStream, positions, "first");
verifyRecord(compressibleInputStream, positions, "third");
verifyRecord(compressibleInputStream, positions, "fifth");
@@ -133,7 +143,7 @@ class CompressibleFSDataInputStreamTest {
new InputStreamFSInputWrapper(new ByteArrayInputStream(compressedBytes));
final FSDataInputStream compressibleInputStream =
new CompressibleFSDataInputStream(
- inputStream, new SnappyStreamCompressionDecorator())) {
+ inputStream, streamCompressionDecorator)) {
verifyRecordPrefix(compressibleInputStream, positions, "first", "fir");
verifyRecordPrefix(compressibleInputStream, positions, "third", "thi");
verifyRecord(compressibleInputStream, positions, "fifth");