This is an automated email from the ASF dual-hosted git repository. aokolnychyi pushed a commit to branch 0.11.x in repository https://gitbox.apache.org/repos/asf/iceberg.git
commit 1c007f29a2b2f4c55d98ab5e809b0d24f05c0396 Author: Jack Ye <[email protected]> AuthorDate: Fri Mar 26 14:28:30 2021 -0700 Core: Support dynamic loading for HadoopFileIO (#2333) --- .../org/apache/iceberg/hadoop/HadoopFileIO.java | 23 ++++++++++++++++++++-- .../java/org/apache/iceberg/TestCatalogUtil.java | 10 ++++++++++ 2 files changed, 31 insertions(+), 2 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 fbaa75d..34d66bf 100644 --- a/core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java +++ b/core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java @@ -20,6 +20,7 @@ package org.apache.iceberg.hadoop; import java.io.IOException; +import org.apache.hadoop.conf.Configurable; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; @@ -29,9 +30,17 @@ import org.apache.iceberg.io.InputFile; import org.apache.iceberg.io.OutputFile; import org.apache.iceberg.util.SerializableSupplier; -public class HadoopFileIO implements FileIO { +public class HadoopFileIO implements FileIO, Configurable { - private final SerializableSupplier<Configuration> hadoopConf; + private SerializableSupplier<Configuration> hadoopConf; + + /** + * Constructor used for dynamic FileIO loading. + * <p> + * {@link Configuration Hadoop configuration} must be set through {@link HadoopFileIO#setConf(Configuration)} + */ + public HadoopFileIO() { + } public HadoopFileIO(Configuration hadoopConf) { this(new SerializableConfiguration(hadoopConf)::get); @@ -65,4 +74,14 @@ public class HadoopFileIO implements FileIO { throw new RuntimeIOException(e, "Failed to delete file: %s", path); } } + + @Override + public void setConf(Configuration conf) { + this.hadoopConf = new SerializableConfiguration(conf)::get; + } + + @Override + public Configuration getConf() { + return hadoopConf.get(); + } } diff --git a/core/src/test/java/org/apache/iceberg/TestCatalogUtil.java b/core/src/test/java/org/apache/iceberg/TestCatalogUtil.java index 37d66c5..c2487eb 100644 --- a/core/src/test/java/org/apache/iceberg/TestCatalogUtil.java +++ b/core/src/test/java/org/apache/iceberg/TestCatalogUtil.java @@ -27,6 +27,7 @@ import org.apache.hadoop.conf.Configuration; import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.hadoop.HadoopFileIO; import org.apache.iceberg.io.FileIO; import org.apache.iceberg.io.InputFile; import org.apache.iceberg.io.OutputFile; @@ -124,6 +125,15 @@ public class TestCatalogUtil { } @Test + public void loadCustomFileIO_hadoopConfigConstructor() { + Configuration configuration = new Configuration(); + configuration.set("key", "val"); + FileIO fileIO = CatalogUtil.loadFileIO(HadoopFileIO.class.getName(), Maps.newHashMap(), configuration); + Assert.assertTrue(fileIO instanceof HadoopFileIO); + Assert.assertEquals("val", ((HadoopFileIO) fileIO).conf().get("key")); + } + + @Test public void loadCustomFileIO_configurable() { Configuration configuration = new Configuration(); configuration.set("key", "val");
