hadoopConf;
-
- /**
- * No-arg constructor to load the FileIO dynamically.
- *
- * All fields are initialized by calling {@link DorisIcebergRestResolvedIO#initialize(Map)} later.
- */
- public DorisIcebergRestResolvedIO() {}
-
- @Override
- public InputFile newInputFile(String location) {
- return io(location).newInputFile(location);
- }
-
- @Override
- public InputFile newInputFile(String location, long length) {
- return io(location).newInputFile(location, length);
- }
-
- @Override
- public OutputFile newOutputFile(String location) {
- return io(location).newOutputFile(location);
- }
-
- @Override
- public void deleteFile(String location) {
- io(location).deleteFile(location);
- }
-
- @Override
- public Map properties() {
- return properties.immutableMap();
- }
-
- @Override
- public void initialize(Map newProperties) {
- close(); // close and discard any existing FileIO instances
- this.properties = SerializableMap.copyOf(newProperties);
- }
-
- @Override
- public void close() {
- List instances = Lists.newArrayList();
-
- synchronized (ioInstances) {
- instances.addAll(ioInstances.values());
- ioInstances.clear();
- }
-
- for (FileIO io : instances) {
- io.close();
- }
- }
-
- @Override
- public void serializeConfWith(
- Function> confSerializer) {
- this.hadoopConf = confSerializer.apply(hadoopConf.get());
- }
-
- @Override
- public void setConf(Configuration conf) {
- this.hadoopConf = new SerializableConfiguration(conf)::get;
- }
-
- @Override
- public Configuration getConf() {
- return hadoopConf.get();
- }
-
- private FileIO io(String location) {
- String impl = implFromLocation(location);
- FileIO io = ioInstances.get(impl);
- if (io != null) {
- return io;
- }
-
- synchronized (ioInstances) {
- // double check while holding the lock
- io = ioInstances.get(impl);
- if (io != null) {
- return io;
- }
-
- Configuration conf = hadoopConf.get();
-
- try {
- if (impl.equals(S3_FILE_IO_IMPL)) {
- io = createS3FileIO(properties);
- } else {
- io = CatalogUtil.loadFileIO(impl, properties, conf);
- }
- } catch (IllegalArgumentException e) {
- if (impl.equals(FALLBACK_IMPL)) {
- // no implementation to fall back to, throw the exception
- throw e;
- } else {
- // couldn't load the normal class, fall back to HadoopFileIO
- LOG.warn(
- "Failed to load FileIO implementation: {}, falling back to {}",
- impl,
- FALLBACK_IMPL,
- e);
- try {
- io = CatalogUtil.loadFileIO(FALLBACK_IMPL, properties, conf);
- } catch (IllegalArgumentException suppressed) {
- LOG.warn(
- "Failed to load FileIO implementation: {} (fallback)", FALLBACK_IMPL, suppressed);
- // both attempts failed, throw the original exception with the later exception
- // suppressed
- e.addSuppressed(suppressed);
- throw e;
- }
- }
- }
-
- ioInstances.put(impl, io);
- }
-
- return io;
- }
-
- private static String implFromLocation(String location) {
- return SCHEME_TO_FILE_IO.getOrDefault(scheme(location), FALLBACK_IMPL);
- }
-
- private static String scheme(String location) {
- int colonPos = location.indexOf(":");
- if (colonPos > 0) {
- return location.substring(0, colonPos);
- }
-
- return null;
- }
-
- protected FileIO createS3FileIO(Map properties) {
-
- // get region
- String region = properties.getOrDefault(S3Properties.REGION,
- properties.getOrDefault(AWSGlueConfig.AWS_REGION, properties.get(Env.REGION)));
-
- // get endpoint
- String s3Endpoint = properties.getOrDefault(S3Properties.ENDPOINT, properties.get(Env.ENDPOINT));
- URI endpointUri = URI.create(s3Endpoint);
-
- // set credential
- CloudCredential credential = new CloudCredential();
- credential.setAccessKey(properties.getOrDefault(S3Properties.ACCESS_KEY,
- properties.get(S3Properties.Env.ACCESS_KEY)));
- credential.setSecretKey(properties.getOrDefault(S3Properties.SECRET_KEY,
- properties.get(S3Properties.Env.SECRET_KEY)));
- if (properties.containsKey(OssProperties.SESSION_TOKEN)
- || properties.containsKey(S3Properties.Env.TOKEN)) {
- credential.setSessionToken(properties.getOrDefault(OssProperties.SESSION_TOKEN,
- properties.get(S3Properties.Env.TOKEN)));
- }
-
- FileIO io = new S3FileIO(() -> S3Util.buildS3Client(endpointUri, region, credential));
- io.initialize(properties);
- return io;
- }
-}
diff --git a/fe/pom.xml b/fe/pom.xml
index 3cb5af10e6e48f..a95a62b2c8d6bf 100644
--- a/fe/pom.xml
+++ b/fe/pom.xml
@@ -302,7 +302,7 @@ under the License.
- 1.1.0
+ 1.4.3
3.0.0rc1
0.45.2-public
1.11.2
@@ -1288,6 +1288,11 @@ under the License.
iceberg-aws
${iceberg.version}