[#5019] feat: (hadoop-catalog): Add a framework to support multi-storage in a pluggable manner for fileset catalog

.gitignore

@@ -53,3 +53,7 @@ include clients/client-python/.gitignore
 **/metastore_db
 **/spark-warehouse
 derby.log
+
+web/node_modules
+web/dist
+web/.next

catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/HadoopCatalogOperations.java

@@ -30,7 +30,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.stream.Collectors;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.gravitino.Catalog;
 import org.apache.gravitino.Entity;
@@ -44,6 +43,8 @@
 import org.apache.gravitino.audit.CallerContext;
 import org.apache.gravitino.audit.FilesetAuditConstants;
 import org.apache.gravitino.audit.FilesetDataOperation;
+import org.apache.gravitino.catalog.hadoop.fs.FileSystemProvider;
+import org.apache.gravitino.catalog.hadoop.fs.FileSystemUtils;
 import org.apache.gravitino.connector.CatalogInfo;
 import org.apache.gravitino.connector.CatalogOperations;
 import org.apache.gravitino.connector.HasPropertyMetadata;
@@ -71,11 +72,10 @@
 import org.slf4j.LoggerFactory;
 
 public class HadoopCatalogOperations implements CatalogOperations, SupportsSchemas, FilesetCatalog {
-
+  private static final String LOCAL_FILE_SCHEME = "file";
   private static final String SCHEMA_DOES_NOT_EXIST_MSG = "Schema %s does not exist";
   private static final String FILESET_DOES_NOT_EXIST_MSG = "Fileset %s does not exist";
   private static final String SLASH = "/";
-
   private static final Logger LOG = LoggerFactory.getLogger(HadoopCatalogOperations.class);
 
   private final EntityStore store;
@@ -90,6 +90,10 @@ public class HadoopCatalogOperations implements CatalogOperations, SupportsSchem
 
   private CatalogInfo catalogInfo;
 
+  private final Map<String, FileSystemProvider> fileSystemProvidersMap = Maps.newHashMap();
+
+  private String defaultFileSystemProviderScheme;
+
   HadoopCatalogOperations(EntityStore store) {
     this.store = store;
   }
@@ -107,7 +111,9 @@ public CatalogInfo getCatalogInfo() {
   }
 
   public Configuration getHadoopConf() {
-    return hadoopConf;
+    Configuration configuration = new Configuration();
+    conf.forEach((k, v) -> configuration.set(k.replace(CATALOG_BYPASS_PREFIX, ""), v));
+    return configuration;
   }
 
   public Map<String, String> getConf() {
@@ -119,26 +125,35 @@ public void initialize(
       Map<String, String> config, CatalogInfo info, HasPropertyMetadata propertiesMetadata)
       throws RuntimeException {
     this.propertiesMetadata = propertiesMetadata;
-    // Initialize Hadoop Configuration.
-    this.conf = config;
-    this.hadoopConf = new Configuration();
     this.catalogInfo = info;
-    Map<String, String> bypassConfigs =
-        config.entrySet().stream()
-            .filter(e -> e.getKey().startsWith(CATALOG_BYPASS_PREFIX))
-            .collect(
-                Collectors.toMap(
-                    e -> e.getKey().substring(CATALOG_BYPASS_PREFIX.length()),
-                    Map.Entry::getValue));
-    bypassConfigs.forEach(hadoopConf::set);
+
+    this.conf = config;
+
+    String fileSystemProviders =
+        (String)
+            propertiesMetadata
+                .catalogPropertiesMetadata()
+                .getOrDefault(
+                    config, HadoopCatalogPropertiesMetadata.FILESYSTEM_PROVIDERS_CLASSNAMES);
+    FileSystemUtils.initFileSystemProviders(fileSystemProviders, fileSystemProvidersMap);
+
+    String defaultFileSystemProviderClassName =
+        (String)
+            propertiesMetadata
+                .catalogPropertiesMetadata()
+                .getOrDefault(
+                    config, HadoopCatalogPropertiesMetadata.DEFAULT_FS_PROVIDER_CLASSNAME);
+    this.defaultFileSystemProviderScheme =
+        StringUtils.isNotBlank(defaultFileSystemProviderClassName)
+            ? FileSystemUtils.getSchemeByFileSystemProvider(
+                defaultFileSystemProviderClassName, fileSystemProvidersMap)
+            : LOCAL_FILE_SCHEME;
 
     String catalogLocation =
         (String)
             propertiesMetadata
                 .catalogPropertiesMetadata()
                 .getOrDefault(config, HadoopCatalogPropertiesMetadata.LOCATION);
-    conf.forEach(hadoopConf::set);
-
     this.catalogStorageLocation =
         StringUtils.isNotBlank(catalogLocation)
             ? Optional.of(catalogLocation).map(Path::new)
@@ -235,8 +250,9 @@ public Fileset createFileset(
 
     try {
       // formalize the path to avoid path without scheme, uri, authority, etc.
-      filesetPath = formalizePath(filesetPath, hadoopConf);
-      FileSystem fs = filesetPath.getFileSystem(hadoopConf);
+      filesetPath = formalizePath(filesetPath, conf);
+
+      FileSystem fs = getFileSystem(filesetPath, conf);
       if (!fs.exists(filesetPath)) {
         if (!fs.mkdirs(filesetPath)) {
           throw new RuntimeException(
@@ -339,7 +355,7 @@ public boolean dropFileset(NameIdentifier ident) {
 
       // For managed fileset, we should delete the related files.
       if (filesetEntity.filesetType() == Fileset.Type.MANAGED) {
-        FileSystem fs = filesetPath.getFileSystem(hadoopConf);
+        FileSystem fs = getFileSystem(filesetPath, conf);
         if (fs.exists(filesetPath)) {
           if (!fs.delete(filesetPath, true)) {
             LOG.warn("Failed to delete fileset {} location {}", ident, filesetPath);
@@ -459,7 +475,7 @@ public Schema createSchema(NameIdentifier ident, String comment, Map<String, Str
     Path schemaPath = getSchemaPath(ident.name(), properties);
     if (schemaPath != null) {
       try {
-        FileSystem fs = schemaPath.getFileSystem(hadoopConf);
+        FileSystem fs = getFileSystem(schemaPath, conf);
         if (!fs.exists(schemaPath)) {
           if (!fs.mkdirs(schemaPath)) {
             // Fail the operation when failed to create the schema path.
@@ -577,8 +593,7 @@ public boolean dropSchema(NameIdentifier ident, boolean cascade) throws NonEmpty
       if (schemaPath == null) {
         return false;
       }
-
-      FileSystem fs = schemaPath.getFileSystem(hadoopConf);
+      FileSystem fs = getFileSystem(schemaPath, conf);
       // Nothing to delete if the schema path does not exist.
       if (!fs.exists(schemaPath)) {
         return false;
@@ -717,8 +732,8 @@ private Path getSchemaPath(String name, Map<String, String> properties) {
   }
 
   @VisibleForTesting
-  static Path formalizePath(Path path, Configuration configuration) throws IOException {
-    FileSystem defaultFs = FileSystem.get(configuration);
+  Path formalizePath(Path path, Map<String, String> configuration) throws IOException {
+    FileSystem defaultFs = getFileSystem(path, configuration);
     return path.makeQualified(defaultFs.getUri(), defaultFs.getWorkingDirectory());
   }
 
@@ -731,7 +746,7 @@ private boolean hasCallerContext() {
   private boolean checkSingleFile(Fileset fileset) {
     try {
       Path locationPath = new Path(fileset.storageLocation());
-      return locationPath.getFileSystem(hadoopConf).getFileStatus(locationPath).isFile();
+      return getFileSystem(locationPath, conf).getFileStatus(locationPath).isFile();
     } catch (FileNotFoundException e) {
       // We should always return false here, same with the logic in `FileSystem.isFile(Path f)`.
       return false;
@@ -742,4 +757,25 @@ private boolean checkSingleFile(Fileset fileset) {
           fileset.name());
     }
   }
+
+  FileSystem getFileSystem(Path path, Map<String, String> config) throws IOException {
+    if (path == null) {
+      throw new IllegalArgumentException("Path should not be null");
+    }
+
+    String scheme =
+        path.toUri().getScheme() != null
+            ? path.toUri().getScheme()
+            : defaultFileSystemProviderScheme;
+
+    FileSystemProvider provider = fileSystemProvidersMap.get(scheme);
+    if (provider == null) {
+      throw new IllegalArgumentException(
+          String.format(
+              "Unsupported scheme: %s, path: %s, all supported scheme: %s",
+              scheme, path, fileSystemProvidersMap.keySet()));
+    }
+
+    return provider.getFileSystem(path, config);
+  }
 }

catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/HadoopCatalogPropertiesMetadata.java

@@ -22,6 +22,8 @@
 import java.util.Map;
 import org.apache.gravitino.catalog.hadoop.authentication.AuthenticationConfig;
 import org.apache.gravitino.catalog.hadoop.authentication.kerberos.KerberosConfig;
+import org.apache.gravitino.catalog.hadoop.fs.FileSystemProvider;
+import org.apache.gravitino.catalog.hadoop.fs.LocalFileSystemProvider;
 import org.apache.gravitino.connector.BaseCatalogPropertiesMetadata;
 import org.apache.gravitino.connector.PropertyEntry;
 
@@ -34,6 +36,21 @@ public class HadoopCatalogPropertiesMetadata extends BaseCatalogPropertiesMetada
   // If not, users have to specify the storage location in the Schema or Fileset level.
   public static final String LOCATION = "location";
 
+  /**
+   * The class names of {@link FileSystemProvider} to be added to the catalog. Except built-in
+   * FileSystemProvider like LocalFileSystemProvider and HDFSFileSystemProvider, users can add their
+   * own FileSystemProvider by specifying the class name here. The value can be
+   * 'xxxx.yyy.FileSystemProvider1,xxxx.yyy.FileSystemProvider2'.
+   */
+  public static final String FILESYSTEM_PROVIDERS_CLASSNAMES = "filesystem-providers-classnames";
+
+  /**
+   * The default file system provider class name, used to create the default file system. If not
+   * specified, the default file system provider will be {@link LocalFileSystemProvider}.
+   */
+  public static final String DEFAULT_FS_PROVIDER_CLASSNAME =
+      "default-filesystem-provider-classname";
+
   private static final Map<String, PropertyEntry<?>> HADOOP_CATALOG_PROPERTY_ENTRIES =
       ImmutableMap.<String, PropertyEntry<?>>builder()
           .put(
@@ -44,6 +61,22 @@ public class HadoopCatalogPropertiesMetadata extends BaseCatalogPropertiesMetada
                   false /* immutable */,
                   null,
                   false /* hidden */))
+          .put(
+              FILESYSTEM_PROVIDERS_CLASSNAMES,
+              PropertyEntry.stringOptionalPropertyEntry(
+                  FILESYSTEM_PROVIDERS_CLASSNAMES,
+                  "The file system provider class name, separated by comma",
+                  false /* immutable */,
+                  null,
+                  false /* hidden */))
+          .put(
+              DEFAULT_FS_PROVIDER_CLASSNAME,
+              PropertyEntry.stringOptionalPropertyEntry(
+                  DEFAULT_FS_PROVIDER_CLASSNAME,
+                  "Default file system provider, used to create the default file system",
+                  false /* immutable */,
+                  LocalFileSystemProvider.class.getCanonicalName(),
+                  false /* hidden */))
           // The following two are about authentication.
           .putAll(KerberosConfig.KERBEROS_PROPERTY_ENTRIES)
           .putAll(AuthenticationConfig.AUTHENTICATION_PROPERTY_ENTRIES)

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

@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.gravitino.catalog.hadoop.fs;
+
+import java.io.IOException;
+import java.util.Map;
+import javax.annotation.Nonnull;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+/**
+ * FileSystemProvider is an interface for providing FileSystem instances. It is used by the
+ * HadoopCatalog to create FileSystem instances for accessing Hadoop compatible file systems.
+ */
+public interface FileSystemProvider {
+
+  /**
+   * Get the FileSystem instance according to the configuration map and file path.
+   *
+   * <p>Compared to the {@link FileSystem#get(Configuration)} method, this method allows the
+   * provider to create a FileSystem instance with a specific configuration and do further
+   * initialization if needed.
+   *
+   * <p>For example: 1. We can check the endpoint value validity for S3AFileSystem then do further
+   * actions. 2. We can also change some default behavior of the FileSystem initialization process
+   * 3. More...
+   *
+   * @param config The configuration for the FileSystem instance.
+   * @param path The path to the file system.
+   * @return The FileSystem instance.
+   * @throws IOException If the FileSystem instance cannot be created.
+   */
+  FileSystem getFileSystem(@Nonnull Path path, @Nonnull Map<String, String> config)
+      throws IOException;
+
+  /**
+   * Get the scheme of this FileSystem provider. The value is 'file' for LocalFileSystem, 'hdfs' for
+   * HDFS, 's3a' for S3AFileSystem, etc.
+   *
+   * @return The scheme of this FileSystem provider.
+   */
+  String getScheme();
+}

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

@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.catalog.hadoop.fs;
+
+import java.util.Map;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.gravitino.exceptions.GravitinoRuntimeException;
+
+public class FileSystemUtils {
+
+  private FileSystemUtils() {}
+
+  public static void initFileSystemProviders(
+      String fileSystemProviders, Map<String, FileSystemProvider> fileProvidersMap) {
+    FileSystemProvider localFileSystemProvider = new LocalFileSystemProvider();
+    FileSystemProvider hdfsFileSystemProvider = new HDFSFileSystemProvider();
+    fileProvidersMap.put(localFileSystemProvider.getScheme(), localFileSystemProvider);
+    fileProvidersMap.put(hdfsFileSystemProvider.getScheme(), hdfsFileSystemProvider);
+
+    if (StringUtils.isBlank(fileSystemProviders)) {
+      return;
+    }
+
+    String[] providers = fileSystemProviders.split(",");
+    for (String provider : providers) {
+      try {
+        FileSystemProvider fileSystemProvider =
+            (FileSystemProvider)
+                Class.forName(provider.trim()).getDeclaredConstructor().newInstance();
+        fileProvidersMap.put(fileSystemProvider.getScheme(), fileSystemProvider);
+      } catch (Exception e) {
+        throw new GravitinoRuntimeException(
+            e, "Failed to initialize file system provider: %s", provider);
+      }
+    }
+  }
+
+  public static String getSchemeByFileSystemProvider(
+      String providerClassName, Map<String, FileSystemProvider> fileProvidersMap) {
+    for (Map.Entry<String, FileSystemProvider> entry : fileProvidersMap.entrySet()) {
+      if (entry.getValue().getClass().getName().equals(providerClassName)) {
+        return entry.getKey();
+      }
+    }
+
+    throw new UnsupportedOperationException(
+        String.format(
+            "File system provider class name '%s' not found. Supported file system providers: %s",
+            providerClassName, fileProvidersMap.values()));
+  }
+}