Cache Hadoop Filesystem instance on Gravitino server to improve the performance

Author: sunxiaojian

bundles/aliyun/src/main/java/org/apache/gravitino/oss/fs/OSSFileSystemProvider.java

@@ -67,7 +67,7 @@ public FileSystem getFileSystem(Path path, Map<String, String> config) throws IO
 
     hadoopConfMap.forEach(configuration::set);
 
-    return AliyunOSSFileSystem.newInstance(path.toUri(), configuration);
+    return AliyunOSSFileSystem.get(path.toUri(), configuration);
   }
 
   @Override

bundles/aws/src/main/java/org/apache/gravitino/s3/fs/S3FileSystemProvider.java

@@ -74,7 +74,7 @@ public FileSystem getFileSystem(Path path, Map<String, String> config) throws IO
     // Hadoop-aws 2 does not support IAMInstanceCredentialsProvider
     checkAndSetCredentialProvider(configuration);
 
-    return S3AFileSystem.newInstance(path.toUri(), configuration);
+    return S3AFileSystem.get(path.toUri(), configuration);
   }
 
   @Override

bundles/azure/src/main/java/org/apache/gravitino/abs/fs/AzureFileSystemProvider.java

@@ -74,7 +74,7 @@ public FileSystem getFileSystem(@Nonnull Path path, @Nonnull Map<String, String>
 
     hadoopConfMap.forEach(configuration::set);
 
-    return FileSystem.newInstance(path.toUri(), configuration);
+    return FileSystem.get(path.toUri(), configuration);
   }
 
   @Override

bundles/gcp/src/main/java/org/apache/gravitino/gcs/fs/GCSFileSystemProvider.java

@@ -49,7 +49,7 @@ public FileSystem getFileSystem(Path path, Map<String, String> config) throws IO
     FileSystemUtils.toHadoopConfigMap(config, GRAVITINO_KEY_TO_GCS_HADOOP_KEY)
         .forEach(configuration::set);
 
-    return FileSystem.newInstance(path.toUri(), configuration);
+    return FileSystem.get(path.toUri(), configuration);
   }
 
   @Override

catalogs/hadoop-common/src/main/java/org/apache/gravitino/catalog/hadoop/fs/FileSystemProvider.java

@@ -65,6 +65,23 @@ public interface FileSystemProvider {
   FileSystem getFileSystem(@Nonnull Path path, @Nonnull Map<String, String> config)
       throws IOException;
 
+  /**
+   * Get the FileSystem instance according to the configuration map and file path.
+   *
+   * @param config The configuration for the FileSystem instance.
+   * @param path The path to the file system.
+   * @param disableCache Whether to cache the FileSystem instance.
+   * @return The FileSystem instance.
+   * @throws IOException If the FileSystem instance cannot be created.
+   */
+  default FileSystem getFileSystem(
+      @Nonnull Path path, @Nonnull Map<String, String> config, boolean disableCache)
+      throws IOException {
+    // disable cache
+    config.put(String.format("fs.%s.impl.disable.cache", scheme()), String.valueOf(disableCache));
+    return getFileSystem(path, config);
+  }
+
   /**
    * Scheme of this FileSystem provider. The value is 'file' for LocalFileSystem, 'hdfs' for HDFS,
    * etc.

catalogs/hadoop-common/src/main/java/org/apache/gravitino/catalog/hadoop/fs/HDFSFileSystemProvider.java

@@ -37,7 +37,7 @@ public FileSystem getFileSystem(@Nonnull Path path, @Nonnull Map<String, String>
         (k, v) -> {
           configuration.set(k.replace(GRAVITINO_BYPASS, ""), v);
         });
-    return FileSystem.newInstance(path.toUri(), configuration);
+    return FileSystem.get(path.toUri(), configuration);
   }
 
   @Override

catalogs/hadoop-common/src/main/java/org/apache/gravitino/catalog/hadoop/fs/LocalFileSystemProvider.java

@@ -37,7 +37,7 @@ public FileSystem getFileSystem(Path path, Map<String, String> config) throws IO
           configuration.set(k.replace(BUILTIN_HDFS_FS_PROVIDER, ""), v);
         });
 
-    return FileSystem.newInstance(path.toUri(), configuration);
+    return FileSystem.get(path.toUri(), configuration);
   }
 
   @Override

clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/GravitinoVirtualFileSystem.java

@@ -319,7 +319,7 @@ private FilesetContextPair getFilesetContext(Path virtualPath, FilesetDataOperat
 
                 totalProperty.putAll(getCredentialProperties(provider, catalog, identifier));
 
-                return provider.getFileSystem(filePath, totalProperty);
+                return provider.getFileSystem(filePath, totalProperty, true);
               } catch (IOException ioe) {
                 throw new GravitinoRuntimeException(
                     ioe,