From 1dfcf7233b74976d1b9fdbcecb1c0516e2cb7e20 Mon Sep 17 00:00:00 2001 From: Steven Wu Date: Mon, 12 Aug 2024 21:06:46 -0700 Subject: [PATCH 1/5] Core: create an empty Hadoop config if not provided in constructor --- .../apache/iceberg/hadoop/HadoopFileIO.java | 8 +++- .../iceberg/hadoop/HadoopFileIOTest.java | 48 +++++++++++++++++++ 2 files changed, 55 insertions(+), 1 deletion(-) diff --git a/core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java b/core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java index 60270749b5d7..343138474eec 100644 --- a/core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java +++ b/core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java @@ -74,7 +74,7 @@ public HadoopFileIO(SerializableSupplier hadoopConf) { } public Configuration conf() { - return hadoopConf.get(); + return getConf(); } @Override @@ -120,6 +120,12 @@ public void setConf(Configuration conf) { @Override public Configuration getConf() { + // Create a default hadoopConf as it is required for the object to be valid. + // E.g. newInputFile would throw NPE with hadoopConf.get() otherwise. + if (hadoopConf == null) { + this.hadoopConf = new SerializableConfiguration(new Configuration())::get; + } + return hadoopConf.get(); } diff --git a/core/src/test/java/org/apache/iceberg/hadoop/HadoopFileIOTest.java b/core/src/test/java/org/apache/iceberg/hadoop/HadoopFileIOTest.java index 1abbd94839de..a90c83475fdd 100644 --- a/core/src/test/java/org/apache/iceberg/hadoop/HadoopFileIOTest.java +++ b/core/src/test/java/org/apache/iceberg/hadoop/HadoopFileIOTest.java @@ -24,6 +24,7 @@ import java.io.File; import java.io.IOException; import java.io.UncheckedIOException; +import java.nio.file.Files; import java.util.List; import java.util.Random; import java.util.UUID; @@ -36,6 +37,7 @@ import org.apache.iceberg.common.DynMethods; import org.apache.iceberg.io.BulkDeletionFailureException; import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.FileIOParser; import org.apache.iceberg.io.ResolvingFileIO; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.Lists; @@ -176,6 +178,52 @@ public void testResolvingFileIOLoad() { assertThat(result).isInstanceOf(HadoopFileIO.class); } + @Test + public void testJsonParserWithoutHadoopConf() throws Exception { + this.hadoopFileIO = new HadoopFileIO(); + + hadoopFileIO.initialize(ImmutableMap.of("properties-bar", "2")); + assertThat(hadoopFileIO.properties().get("properties-bar")).isEqualTo("2"); + + testJsonParser(hadoopFileIO, tempDir); + } + + @Test + public void testJsonParserWithHadoopConf() throws Exception { + this.hadoopFileIO = new HadoopFileIO(); + + Configuration hadoopConf = new Configuration(); + hadoopConf.setInt("hadoop-conf-foo", 1); + hadoopFileIO.setConf(hadoopConf); + assertThat(hadoopFileIO.conf().get("hadoop-conf-foo")).isNotNull(); + + hadoopFileIO.initialize(ImmutableMap.of("properties-bar", "2")); + assertThat(hadoopFileIO.properties().get("properties-bar")).isEqualTo("2"); + + testJsonParser(hadoopFileIO, tempDir); + } + + private static void testJsonParser(HadoopFileIO hadoopFileIO, File tempDir) throws Exception { + String json = FileIOParser.toJson(hadoopFileIO); + try (FileIO deserialized = FileIOParser.fromJson(json)) { + assertThat(deserialized).isInstanceOf(HadoopFileIO.class); + HadoopFileIO deserializedHadoopFileIO = (HadoopFileIO) deserialized; + + // properties are carried over during serialization and deserialization + assertThat(deserializedHadoopFileIO.properties()).isEqualTo(hadoopFileIO.properties()); + + // FileIOParser doesn't serialize and deserialize Hadoop configuration + // so config "foo" is not restored in deserialized object. + assertThat(deserializedHadoopFileIO.conf().get("hadoop-conf-foo")).isNull(); + + // make sure deserialized io can create input file + String inputFilePath = + Files.createTempDirectory(tempDir.toPath(), "junit").toFile().getAbsolutePath() + + "/test.parquet"; + deserializedHadoopFileIO.newInputFile(inputFilePath); + } + } + private List createRandomFiles(Path parent, int count) { Vector paths = new Vector<>(); random From 55dd00b1eb640aaf5d4609ae4d991afc9dfa2809 Mon Sep 17 00:00:00 2001 From: Steven Wu Date: Mon, 14 Oct 2024 09:43:42 -0700 Subject: [PATCH 2/5] switch all `hadoopConf.get()` to `getConf()` --- .../org/apache/iceberg/hadoop/HadoopFileIO.java | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java b/core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java index 343138474eec..88a2bd50a3e9 100644 --- a/core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java +++ b/core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java @@ -84,23 +84,23 @@ public void initialize(Map props) { @Override public InputFile newInputFile(String path) { - return HadoopInputFile.fromLocation(path, hadoopConf.get()); + return HadoopInputFile.fromLocation(path, getConf()); } @Override public InputFile newInputFile(String path, long length) { - return HadoopInputFile.fromLocation(path, length, hadoopConf.get()); + return HadoopInputFile.fromLocation(path, length, getConf()); } @Override public OutputFile newOutputFile(String path) { - return HadoopOutputFile.fromPath(new Path(path), hadoopConf.get()); + return HadoopOutputFile.fromPath(new Path(path), getConf()); } @Override public void deleteFile(String path) { Path toDelete = new Path(path); - FileSystem fs = Util.getFs(toDelete, hadoopConf.get()); + FileSystem fs = Util.getFs(toDelete, getConf()); try { fs.delete(toDelete, false /* not recursive */); } catch (IOException e) { @@ -121,7 +121,7 @@ public void setConf(Configuration conf) { @Override public Configuration getConf() { // Create a default hadoopConf as it is required for the object to be valid. - // E.g. newInputFile would throw NPE with hadoopConf.get() otherwise. + // E.g. newInputFile would throw NPE with getConf() otherwise. if (hadoopConf == null) { this.hadoopConf = new SerializableConfiguration(new Configuration())::get; } @@ -138,7 +138,7 @@ public void serializeConfWith( @Override public Iterable listPrefix(String prefix) { Path prefixToList = new Path(prefix); - FileSystem fs = Util.getFs(prefixToList, hadoopConf.get()); + FileSystem fs = Util.getFs(prefixToList, getConf()); return () -> { try { @@ -160,7 +160,7 @@ public Iterable listPrefix(String prefix) { @Override public void deletePrefix(String prefix) { Path prefixToDelete = new Path(prefix); - FileSystem fs = Util.getFs(prefixToDelete, hadoopConf.get()); + FileSystem fs = Util.getFs(prefixToDelete, getConf()); try { fs.delete(prefixToDelete, true /* recursive */); From e360aa211554fa170acecb1b5b13dc92018d5abc Mon Sep 17 00:00:00 2001 From: Steven Zhen Wu Date: Tue, 15 Oct 2024 10:16:47 -0700 Subject: [PATCH 3/5] Update core/src/test/java/org/apache/iceberg/hadoop/HadoopFileIOTest.java Co-authored-by: Eduard Tudenhoefner --- .../test/java/org/apache/iceberg/hadoop/HadoopFileIOTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/test/java/org/apache/iceberg/hadoop/HadoopFileIOTest.java b/core/src/test/java/org/apache/iceberg/hadoop/HadoopFileIOTest.java index a90c83475fdd..6a29d54ccc3f 100644 --- a/core/src/test/java/org/apache/iceberg/hadoop/HadoopFileIOTest.java +++ b/core/src/test/java/org/apache/iceberg/hadoop/HadoopFileIOTest.java @@ -220,7 +220,7 @@ private static void testJsonParser(HadoopFileIO hadoopFileIO, File tempDir) thro String inputFilePath = Files.createTempDirectory(tempDir.toPath(), "junit").toFile().getAbsolutePath() + "/test.parquet"; - deserializedHadoopFileIO.newInputFile(inputFilePath); + deserializedHadoopFileIO.newInputFile(File.createTempFile("test", "parquet", tempDir).toString()); } } From 3797a97c1a990057c8a8d1b95c156dc687115d47 Mon Sep 17 00:00:00 2001 From: Steven Wu Date: Tue, 15 Oct 2024 13:39:09 -0700 Subject: [PATCH 4/5] fix style --- .../test/java/org/apache/iceberg/hadoop/HadoopFileIOTest.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/core/src/test/java/org/apache/iceberg/hadoop/HadoopFileIOTest.java b/core/src/test/java/org/apache/iceberg/hadoop/HadoopFileIOTest.java index 6a29d54ccc3f..0ff1d98f27dd 100644 --- a/core/src/test/java/org/apache/iceberg/hadoop/HadoopFileIOTest.java +++ b/core/src/test/java/org/apache/iceberg/hadoop/HadoopFileIOTest.java @@ -220,7 +220,8 @@ private static void testJsonParser(HadoopFileIO hadoopFileIO, File tempDir) thro String inputFilePath = Files.createTempDirectory(tempDir.toPath(), "junit").toFile().getAbsolutePath() + "/test.parquet"; - deserializedHadoopFileIO.newInputFile(File.createTempFile("test", "parquet", tempDir).toString()); + deserializedHadoopFileIO.newInputFile( + File.createTempFile("test", "parquet", tempDir).toString()); } } From edda652af6abff442f97be253ed92f9d6f9c00b2 Mon Sep 17 00:00:00 2001 From: Steven Wu Date: Tue, 15 Oct 2024 13:44:18 -0700 Subject: [PATCH 5/5] use double checked lock to lazily initialize hadoopConf --- .../main/java/org/apache/iceberg/hadoop/HadoopFileIO.java | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java b/core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java index 88a2bd50a3e9..02530cf6c9dd 100644 --- a/core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java +++ b/core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java @@ -123,7 +123,11 @@ public Configuration getConf() { // Create a default hadoopConf as it is required for the object to be valid. // E.g. newInputFile would throw NPE with getConf() otherwise. if (hadoopConf == null) { - this.hadoopConf = new SerializableConfiguration(new Configuration())::get; + synchronized (this) { + if (hadoopConf == null) { + this.hadoopConf = new SerializableConfiguration(new Configuration())::get; + } + } } return hadoopConf.get();