From 10ef09c343e149b9a13dc43ff47239de9222026c Mon Sep 17 00:00:00 2001
From: ranxianglei <ranxianglei.rxl@taobao.com>
Date: Tue, 12 Nov 2024 04:01:45 +0000
Subject: [PATCH] [format] recover HadoopFileIO

---
 .../apache/paimon/fs/hadoop/HadoopFileIO.java | 27 ++++++++++++++++++-
 1 file changed, 26 insertions(+), 1 deletion(-)

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 6b8104f1b6da..c948536660f5 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
@@ -27,6 +27,7 @@
 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;
@@ -39,6 +40,9 @@
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.nio.charset.StandardCharsets;
+import java.net.URI;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicReference;
 
 /** Hadoop {@link FileIO}. */
@@ -48,6 +52,8 @@ public class HadoopFileIO implements FileIO {
 
     protected SerializableConfiguration hadoopConf;
 
+    protected transient volatile Map<Pair<String, String>, FileSystem> fsMap;
+
     @VisibleForTesting
     public void setFileSystem(Path path, FileSystem fs) throws IOException {
         org.apache.hadoop.fs.Path hadoopPath = path(path);
@@ -143,7 +149,26 @@ private FileSystem getFileSystem(
             org.apache.hadoop.fs.Path path,
             FunctionWithException<org.apache.hadoop.fs.Path, FileSystem, IOException> creator)
             throws IOException {
-        return creator.apply(path);
+        if (fsMap == null) {
+            synchronized (this) {
+                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;
     }
 
     protected FileSystem createFileSystem(org.apache.hadoop.fs.Path path) throws IOException {