[#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,9 @@
 import org.slf4j.LoggerFactory;
 
 public class HadoopCatalogOperations implements CatalogOperations, SupportsSchemas, FilesetCatalog {
-
   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 +89,10 @@ public class HadoopCatalogOperations implements CatalogOperations, SupportsSchem
 
   private CatalogInfo catalogInfo;
 
+  private final Map<String, FileSystemProvider> fileSystemProvidersMap = Maps.newHashMap();
+
+  private FileSystemProvider defaultFileSystemProvider;
+
   HadoopCatalogOperations(EntityStore store) {
     this.store = store;
   }
@@ -107,7 +110,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 +124,31 @@ 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);
+    this.fileSystemProvidersMap.putAll(FileSystemUtils.getFileSystemProviders(fileSystemProviders));
+
+    String defaultFileSystemProviderName =
+        (String)
+            propertiesMetadata
+                .catalogPropertiesMetadata()
+                .getOrDefault(config, HadoopCatalogPropertiesMetadata.DEFAULT_FS_PROVIDER);
+    this.defaultFileSystemProvider =
+        FileSystemUtils.getFileSystemProviderByName(
+            fileSystemProvidersMap, defaultFileSystemProviderName);
 
     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 +245,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 +350,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 +470,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 +588,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 +727,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 +741,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 +752,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()
+            : defaultFileSystemProvider.scheme();
+
+    FileSystemProvider provider = fileSystemProvidersMap.get(scheme);
+    if (provider == null) {
+      throw new IllegalArgumentException(
+          String.format(
+              "Unsupported scheme: %s, path: %s, all supported scheme: %s and provider: %s",
+              scheme, path, fileSystemProvidersMap.keySet(), fileSystemProvidersMap.values()));
+    }
+
+    return provider.getFileSystem(path, config);
+  }
 }

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

@@ -18,10 +18,13 @@
  */
 package org.apache.gravitino.catalog.hadoop;
 
+import static org.apache.gravitino.catalog.hadoop.authentication.kerberos.KerberosConfig.KERBEROS_PROPERTY_ENTRIES;
+
 import com.google.common.collect.ImmutableMap;
 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 +37,20 @@ 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 name 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 provider name here. The value can be find {@link
+   * FileSystemProvider#name()}.
+   */
+  public static final String FILESYSTEM_PROVIDERS = "filesystem-providers";
+
+  /**
+   * 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#name()}.
+   */
+  public static final String DEFAULT_FS_PROVIDER = "default-filesystem-provider";
+
   private static final Map<String, PropertyEntry<?>> HADOOP_CATALOG_PROPERTY_ENTRIES =
       ImmutableMap.<String, PropertyEntry<?>>builder()
           .put(
@@ -44,8 +61,24 @@ public class HadoopCatalogPropertiesMetadata extends BaseCatalogPropertiesMetada
                   false /* immutable */,
                   null,
                   false /* hidden */))
+          .put(
+              FILESYSTEM_PROVIDERS,
+              PropertyEntry.stringOptionalPropertyEntry(
+                  FILESYSTEM_PROVIDERS,
+                  "The file system provider names, separated by comma",
+                  false /* immutable */,
+                  null,
+                  false /* hidden */))
+          .put(
+              DEFAULT_FS_PROVIDER,
+              PropertyEntry.stringOptionalPropertyEntry(
+                  DEFAULT_FS_PROVIDER,
+                  "Default file system provider name",
+                  false /* immutable */,
+                  LocalFileSystemProvider.class.getSimpleName(),
+                  false /* hidden */))
           // The following two are about authentication.
-          .putAll(KerberosConfig.KERBEROS_PROPERTY_ENTRIES)
+          .putAll(KERBEROS_PROPERTY_ENTRIES)
           .putAll(AuthenticationConfig.AUTHENTICATION_PROPERTY_ENTRIES)
           .build();
 

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

@@ -0,0 +1,69 @@
+/*
+ * 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;
+
+  /**
+   * Scheme of this FileSystem provider. The value is 'file' for LocalFileSystem, 'hdfs' for HDFS,
+   * etc.
+   *
+   * @return The scheme of this FileSystem provider used.
+   */
+  String scheme();
+
+  /**
+   * Name of this FileSystem provider. The value is 'LocalFileSystemProvider' for LocalFileSystem,
+   * 'HDFSFileSystemProvider' for HDFS, etc.
+   *
+   * @return The name of this FileSystem provider.
+   */
+  String name();
+}

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

@@ -0,0 +1,74 @@
+/*
+ * 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 com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.ServiceLoader;
+import java.util.Set;
+
+public class FileSystemUtils {
+
+  private FileSystemUtils() {}
+
+  public static Map<String, FileSystemProvider> getFileSystemProviders(String fileSystemProviders) {
+    Map<String, FileSystemProvider> resultMap = Maps.newHashMap();
+    ServiceLoader<FileSystemProvider> allFileSystemProviders =
+        ServiceLoader.load(FileSystemProvider.class);
+
+    Set<String> providersInUses =
+        fileSystemProviders != null
+            ? Arrays.stream(fileSystemProviders.split(","))
+                .map(String::trim)
+                .collect(java.util.stream.Collectors.toSet())
+            : Sets.newHashSet();
+
+    // Always add the built-in LocalFileSystemProvider and HDFSFileSystemProvider to the catalog.
+    providersInUses.add(LocalFileSystemProvider.class.getSimpleName());
+    providersInUses.add(HDFSFileSystemProvider.class.getSimpleName());
+
+    allFileSystemProviders.forEach(
+        fileSystemProvider -> {
+          if (providersInUses.contains(fileSystemProvider.getClass().getSimpleName())) {
+            if (resultMap.containsKey(fileSystemProvider.scheme())) {
+              throw new UnsupportedOperationException(
+                  String.format(
+                      "File system provider with scheme '%s' already exists in the use provider list "
+                          + "Please make sure the file system provider scheme is unique.",
+                      fileSystemProvider.name()));
+            }
+
+            resultMap.put(fileSystemProvider.scheme(), fileSystemProvider);
+          }
+        });
+
+    return resultMap;
+  }
+
+  public static FileSystemProvider getFileSystemProviderByName(
+      Map<String, FileSystemProvider> fileSystemProviders, String defaultFileSystemProvider) {
+    return fileSystemProviders.entrySet().stream()
+        .filter(entry -> entry.getValue().name().equals(defaultFileSystemProvider))
+        .map(Map.Entry::getValue)
+        .findFirst()
+        .orElse(null);
+  }
+}