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..02530cf6c9dd 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 @@ -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) { @@ -120,6 +120,16 @@ 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 getConf() otherwise. + if (hadoopConf == null) { + synchronized (this) { + if (hadoopConf == null) { + this.hadoopConf = new SerializableConfiguration(new Configuration())::get; + } + } + } + return hadoopConf.get(); } @@ -132,7 +142,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 { @@ -154,7 +164,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 */); 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..0ff1d98f27dd 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,53 @@ 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( + File.createTempFile("test", "parquet", tempDir).toString()); + } + } + private List createRandomFiles(Path parent, int count) { Vector paths = new Vector<>(); random