This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b75338e8e794feb37ea8c3cd1d1b8c2f12d58f72
Author: Etienne Chauchot <echauc...@apache.org>
AuthorDate: Fri Sep 8 11:31:18 2023 +0200
[FLINK-33059][connectors][format][filesystem] Generalize compression test
to all compression formats
---
.../flink/api/common/io/FileInputFormat.java | 4 +++
.../flink/api/common/io/FileInputFormatTest.java | 32 ++++++++++++----------
.../org/apache/flink/testutils/TestFileUtils.java | 21 ++++++++++++++
3 files changed, 42 insertions(+), 15 deletions(-)
diff --git
a/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java
b/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java
index 3a3ad24c9df..6a2a78e6cc2 100644
---
a/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java
+++
b/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java
@@ -157,6 +157,10 @@ public abstract class FileInputFormat<OT> extends
RichInputFormat<OT, FileInputS
}
}
+ public static Set<String> getSupportedCompressionFormats() {
+ return INFLATER_INPUT_STREAM_FACTORIES.keySet();
+ }
+
/**
* Returns the extension of a file name (!= a path).
*
diff --git
a/flink-core/src/test/java/org/apache/flink/api/common/io/FileInputFormatTest.java
b/flink-core/src/test/java/org/apache/flink/api/common/io/FileInputFormatTest.java
index 8b233c85cca..09b4607c548 100644
---
a/flink-core/src/test/java/org/apache/flink/api/common/io/FileInputFormatTest.java
+++
b/flink-core/src/test/java/org/apache/flink/api/common/io/FileInputFormatTest.java
@@ -18,6 +18,8 @@
package org.apache.flink.api.common.io;
+import java.util.Set;
+
import org.apache.flink.api.common.io.FileInputFormat.FileBaseStatistics;
import org.apache.flink.api.common.io.statistics.BaseStatistics;
import org.apache.flink.configuration.ConfigConstants;
@@ -592,31 +594,29 @@ public class FileInputFormatTest {
// Unsplittable input files
// ------------------------------------------------------------------------
- // ---- Tests for .deflate ---------
+ // ---- Tests for compressed files ---------
/**
* Create directory with files with .deflate extension and see if it
creates a split for each
* file. Each split has to start from the beginning.
*/
@Test
- public void testFileInputSplit() {
+ public void testFileInputFormatWithCompression() {
try {
String tempFile =
- TestFileUtils.createTempFileDirExtension(
- temporaryFolder.newFolder(),
- ".deflate",
- "some",
- "stupid",
- "meaningless",
- "files");
+ TestFileUtils.createTempTextFileDirCompressionFormats(
+ temporaryFolder.newFolder());
final DummyFileInputFormat format = new DummyFileInputFormat();
format.setFilePath(tempFile);
format.configure(new Configuration());
FileInputSplit[] splits = format.createInputSplits(2);
- Assert.assertEquals(4, splits.length);
+ final Set<String> supportedCompressionFormats =
+ FileInputFormat.getSupportedCompressionFormats();
+ Assert.assertEquals(supportedCompressionFormats.size(),
splits.length);
for (FileInputSplit split : splits) {
Assert.assertEquals(
- -1L, split.getLength()); // unsplittable deflate files
have this size as a
+ FileInputFormat.READ_WHOLE_SPLIT_FLAG,
+ split.getLength()); // unsplittable compressed files
have this size as a
// flag for "read whole file"
Assert.assertEquals(0L, split.getStart()); // always read from
the beginning.
}
@@ -630,12 +630,14 @@ public class FileInputFormatTest {
formatMixed.setFilePath(tempFile);
formatMixed.configure(new Configuration());
FileInputSplit[] splitsMixed = formatMixed.createInputSplits(2);
- Assert.assertEquals(5, splitsMixed.length);
+ Assert.assertEquals(supportedCompressionFormats.size() + 1,
splitsMixed.length);
for (FileInputSplit split : splitsMixed) {
- if (split.getPath().getName().endsWith(".deflate")) {
+ final String extension =
+
FileInputFormat.extractFileExtension(split.getPath().getName());
+ if (supportedCompressionFormats.contains(extension)) {
Assert.assertEquals(
- -1L,
- split.getLength()); // unsplittable deflate files
have this size as a
+ FileInputFormat.READ_WHOLE_SPLIT_FLAG,
+ split.getLength()); // unsplittable compressed
files have this size as a
// flag for "read whole file"
Assert.assertEquals(0L, split.getStart()); // always read
from the beginning.
} else {
diff --git
a/flink-core/src/test/java/org/apache/flink/testutils/TestFileUtils.java
b/flink-core/src/test/java/org/apache/flink/testutils/TestFileUtils.java
index 71c3825416f..9dedf0cc37c 100644
--- a/flink-core/src/test/java/org/apache/flink/testutils/TestFileUtils.java
+++ b/flink-core/src/test/java/org/apache/flink/testutils/TestFileUtils.java
@@ -26,6 +26,8 @@ import java.io.FileWriter;
import java.io.IOException;
import java.nio.file.Files;
+import org.apache.flink.api.common.io.FileInputFormat;
+
public class TestFileUtils {
private static final String FILE_PREFIX = "flink_test_";
@@ -136,6 +138,25 @@ public class TestFileUtils {
return f.toURI().toString();
}
+ public static String createTempTextFileDirCompressionFormats(File tempDir)
throws IOException {
+ File f = null;
+ do {
+ f = new File(tempDir, randomFileName(FILE_SUFFIX));
+ } while (f.exists());
+ f.mkdirs();
+ f.deleteOnExit();
+
+ for (String extension :
FileInputFormat.getSupportedCompressionFormats()) {
+ File child = new File(f, randomFileName("." + extension));
+ child.deleteOnExit();
+
+ try (BufferedWriter out = new BufferedWriter(new
FileWriter(child))) {
+ out.write("random text " + Math.random());
+ }
+ }
+ return f.toURI().toString();
+ }
+
public static String randomFileName() {
return randomFileName(FILE_SUFFIX);
}
