This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new ce06d0c97 [fs] HadoopFileIO should cache FileSystem by scheme and
authority (#2504)
ce06d0c97 is described below
commit ce06d0c978db70ff469f525e3a3c6bf3708eb1d9
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Dec 13 18:21:29 2023 +0800
[fs] HadoopFileIO should cache FileSystem by scheme and authority (#2504)
---
.../org/apache/paimon/fs/hadoop/HadoopFileIO.java | 38 ++++++++++++++++++----
.../paimon/fs/HadoopLocalFileIOBehaviorTest.java | 2 +-
.../org/apache/paimon/fs/HdfsBehaviorTest.java | 4 +--
.../apache/paimon/oss/HadoopCompliantFileIO.java | 3 ++
.../apache/paimon/s3/HadoopCompliantFileIO.java | 22 ++++++++++---
5 files changed, 56 insertions(+), 13 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java
b/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java
index 7074341a7..70325ee69 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java
@@ -26,6 +26,8 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.hadoop.SerializableConfiguration;
+import org.apache.paimon.utils.FunctionWithException;
+import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.ReflectionUtils;
import org.apache.hadoop.fs.FSDataInputStream;
@@ -37,7 +39,10 @@ import java.io.IOException;
import java.io.OutputStreamWriter;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
+import java.net.URI;
import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;
/** Hadoop {@link FileIO}. */
@@ -47,11 +52,12 @@ public class HadoopFileIO implements FileIO {
protected SerializableConfiguration hadoopConf;
- protected transient volatile FileSystem fs;
+ protected transient volatile Map<Pair<String, String>, FileSystem> fsMap;
@VisibleForTesting
- public void setFileSystem(FileSystem fs) {
- this.fs = fs;
+ public void setFileSystem(Path path, FileSystem fs) throws IOException {
+ org.apache.hadoop.fs.Path hadoopPath = path(path);
+ getFileSystem(hadoopPath, p -> fs);
}
@Override
@@ -136,13 +142,32 @@ public class HadoopFileIO implements FileIO {
}
private FileSystem getFileSystem(org.apache.hadoop.fs.Path path) throws
IOException {
- if (fs == null) {
+ return getFileSystem(path, this::createFileSystem);
+ }
+
+ private FileSystem getFileSystem(
+ org.apache.hadoop.fs.Path path,
+ FunctionWithException<org.apache.hadoop.fs.Path, FileSystem,
IOException> creator)
+ throws IOException {
+ if (fsMap == null) {
synchronized (this) {
- if (fs == null) {
- fs = createFileSystem(path);
+ if (fsMap == null) {
+ fsMap = new ConcurrentHashMap<>();
}
}
}
+
+ Map<Pair<String, String>, FileSystem> map = fsMap;
+
+ URI uri = path.toUri();
+ String scheme = uri.getScheme();
+ String authority = uri.getAuthority();
+ Pair<String, String> key = Pair.of(scheme, authority);
+ FileSystem fs = map.get(key);
+ if (fs == null) {
+ fs = creator.apply(path);
+ map.put(key, fs);
+ }
return fs;
}
@@ -373,6 +398,7 @@ public class HadoopFileIO implements FileIO {
new org.apache.hadoop.fs.Path(
path.getParent(), String.format(".%s.crc",
path.getName()));
+ FileSystem fs = getFileSystem(checksumFile);
if (fs.exists(checksumFile)) {
// checksum file exists, deleting it
fs.delete(checksumFile, true); // recursive=true
diff --git
a/paimon-common/src/test/java/org/apache/paimon/fs/HadoopLocalFileIOBehaviorTest.java
b/paimon-common/src/test/java/org/apache/paimon/fs/HadoopLocalFileIOBehaviorTest.java
index 57a9aba7f..df94dd2a8 100644
---
a/paimon-common/src/test/java/org/apache/paimon/fs/HadoopLocalFileIOBehaviorTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/fs/HadoopLocalFileIOBehaviorTest.java
@@ -39,7 +39,7 @@ class HadoopLocalFileIOBehaviorTest extends
FileIOBehaviorTestBase {
org.apache.hadoop.fs.FileSystem fs = new RawLocalFileSystem();
fs.initialize(URI.create("file:///"), new Configuration());
HadoopFileIO fileIO = new HadoopFileIO();
- fileIO.setFileSystem(fs);
+ fileIO.setFileSystem(getBasePath(), fs);
return fileIO;
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/fs/HdfsBehaviorTest.java
b/paimon-common/src/test/java/org/apache/paimon/fs/HdfsBehaviorTest.java
index df5eb7ba7..56bfe052e 100644
--- a/paimon-common/src/test/java/org/apache/paimon/fs/HdfsBehaviorTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/fs/HdfsBehaviorTest.java
@@ -60,10 +60,10 @@ class HdfsBehaviorTest extends FileIOBehaviorTestBase {
hdfsCluster = builder.build();
org.apache.hadoop.fs.FileSystem hdfs = hdfsCluster.getFileSystem();
- fs = new HadoopFileIO();
- fs.setFileSystem(hdfs);
basePath = new Path(hdfs.getUri().toString() + "/tests");
+ fs = new HadoopFileIO();
+ fs.setFileSystem(basePath, hdfs);
}
@AfterAll
diff --git
a/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/HadoopCompliantFileIO.java
b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/HadoopCompliantFileIO.java
index 3f40c577a..4d86c12a6 100644
---
a/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/HadoopCompliantFileIO.java
+++
b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/HadoopCompliantFileIO.java
@@ -118,6 +118,9 @@ public abstract class HadoopCompliantFileIO implements
FileIO {
Map<String, FileSystem> map = fsMap;
String authority = path.toUri().getAuthority();
+ if (authority == null) {
+ authority = "DEFAULT";
+ }
FileSystem fs = map.get(authority);
if (fs == null) {
fs = createFileSystem(path);
diff --git
a/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/HadoopCompliantFileIO.java
b/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/HadoopCompliantFileIO.java
index fe86aca96..abfe0fabb 100644
---
a/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/HadoopCompliantFileIO.java
+++
b/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/HadoopCompliantFileIO.java
@@ -29,6 +29,8 @@ import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
/**
* Hadoop {@link FileIO}.
@@ -39,7 +41,7 @@ public abstract class HadoopCompliantFileIO implements FileIO
{
private static final long serialVersionUID = 1L;
- protected transient volatile FileSystem fs;
+ protected transient volatile Map<String, FileSystem> fsMap;
@Override
public SeekableInputStream newInputStream(Path path) throws IOException {
@@ -105,13 +107,25 @@ public abstract class HadoopCompliantFileIO implements
FileIO {
}
private FileSystem getFileSystem(org.apache.hadoop.fs.Path path) throws
IOException {
- if (fs == null) {
+ if (fsMap == null) {
synchronized (this) {
- if (fs == null) {
- fs = createFileSystem(path);
+ if (fsMap == null) {
+ fsMap = new ConcurrentHashMap<>();
}
}
}
+
+ Map<String, FileSystem> map = fsMap;
+
+ String authority = path.toUri().getAuthority();
+ if (authority == null) {
+ authority = "DEFAULT";
+ }
+ FileSystem fs = map.get(authority);
+ if (fs == null) {
+ fs = createFileSystem(path);
+ map.put(authority, fs);
+ }
return fs;
}
