[#5220] improvment(hadoop-catalog): Optimize the name properties keys for Hadoop catalog.

bundles/aliyun-bundle/build.gradle.kts

@@ -35,6 +35,9 @@ dependencies {
   // org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem.initialize(AliyunOSSFileSystem.java:323)
   // org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:3611)
   implementation(libs.commons.lang)
+  implementation(project(":catalogs:catalog-common")) {
+    exclude("*")
+  }
 }
 
 tasks.withType(ShadowJar::class.java) {

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

@@ -18,23 +18,40 @@
  */
 package org.apache.gravitino.oss.fs;
 
+import com.google.common.collect.ImmutableMap;
 import java.io.IOException;
 import java.util.Map;
 import org.apache.gravitino.catalog.hadoop.fs.FileSystemProvider;
+import org.apache.gravitino.catalog.hadoop.fs.FileSystemUtils;
+import org.apache.gravitino.storage.OSSProperties;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem;
+import org.apache.hadoop.fs.aliyun.oss.Constants;
 
 public class OSSFileSystemProvider implements FileSystemProvider {
+
+  private static final String OSS_FILESYSTEM_IMPL = "fs.oss.impl";
+
+  private static final Map<String, String> GRAVITINO_KEY_TO_OSS_HADOOP_KEY =
+      ImmutableMap.of(
+          OSSProperties.GRAVITINO_OSS_ENDPOINT, Constants.ENDPOINT_KEY,
+          OSSProperties.GRAVITINO_OSS_ACCESS_KEY_ID, Constants.ACCESS_KEY_ID,
+          OSSProperties.GRAVITINO_OSS_ACCESS_KEY_SECRET, Constants.ACCESS_KEY_SECRET);
+
   @Override
   public FileSystem getFileSystem(Path path, Map<String, String> config) throws IOException {
     Configuration configuration = new Configuration();
-    config.forEach(
-        (k, v) -> {
-          configuration.set(k.replace("gravitino.bypass.", ""), v);
-        });
 
+    Map<String, String> hadoopConfMap =
+        FileSystemUtils.toHadoopConfigMap(config, GRAVITINO_KEY_TO_OSS_HADOOP_KEY);
+    // OSS do not use service loader to load the file system, so we need to set the impl class
+    if (!hadoopConfMap.containsKey(OSS_FILESYSTEM_IMPL)) {
+      hadoopConfMap.put(OSS_FILESYSTEM_IMPL, AliyunOSSFileSystem.class.getCanonicalName());
+    }
+
+    hadoopConfMap.forEach(configuration::set);
     return AliyunOSSFileSystem.newInstance(path.toUri(), configuration);
   }
 

bundles/aws-bundle/build.gradle.kts

@@ -28,6 +28,9 @@ dependencies {
   compileOnly(project(":catalogs:catalog-hadoop"))
   compileOnly(libs.hadoop3.common)
   implementation(libs.hadoop3.aws)
+  implementation(project(":catalogs:catalog-common")) {
+    exclude("*")
+  }
 }
 
 tasks.withType(ShadowJar::class.java) {

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

@@ -19,23 +19,37 @@
 
 package org.apache.gravitino.s3.fs;
 
+import com.google.common.collect.ImmutableMap;
 import java.io.IOException;
 import java.util.Map;
 import org.apache.gravitino.catalog.hadoop.fs.FileSystemProvider;
+import org.apache.gravitino.catalog.hadoop.fs.FileSystemUtils;
+import org.apache.gravitino.storage.S3Properties;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.s3a.Constants;
 import org.apache.hadoop.fs.s3a.S3AFileSystem;
 
 public class S3FileSystemProvider implements FileSystemProvider {
+
+  private static final Map<String, String> GRAVITINO_KEY_TO_S3_HADOOP_KEY =
+      ImmutableMap.of(
+          S3Properties.GRAVITINO_S3_ENDPOINT, Constants.ENDPOINT,
+          S3Properties.GRAVITINO_S3_ACCESS_KEY_ID, Constants.ACCESS_KEY,
+          S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY, Constants.SECRET_KEY);
+
   @Override
   public FileSystem getFileSystem(Path path, Map<String, String> config) throws IOException {
     Configuration configuration = new Configuration();
-    config.forEach(
-        (k, v) -> {
-          configuration.set(k.replace("gravitino.bypass.", ""), v);
-        });
+    Map<String, String> hadoopConfMap =
+        FileSystemUtils.toHadoopConfigMap(config, GRAVITINO_KEY_TO_S3_HADOOP_KEY);
 
+    if (!hadoopConfMap.containsKey(Constants.AWS_CREDENTIALS_PROVIDER)) {
+      configuration.set(
+          Constants.AWS_CREDENTIALS_PROVIDER, Constants.ASSUMED_ROLE_CREDENTIALS_DEFAULT);
+    }
+    hadoopConfMap.forEach(configuration::set);
     return S3AFileSystem.newInstance(path.toUri(), configuration);
   }
 

bundles/gcp-bundle/build.gradle.kts

@@ -34,6 +34,9 @@ dependencies {
 
   implementation(libs.commons.lang3)
   implementation(libs.hadoop3.gcs)
+  implementation(project(":catalogs:catalog-common")) {
+    exclude("*")
+  }
   implementation(libs.google.auth.http)
   implementation(libs.google.auth.credentials)
 }

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

@@ -19,9 +19,12 @@
 package org.apache.gravitino.gcs.fs;
 
 import com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem;
+import com.google.common.collect.ImmutableMap;
 import java.io.IOException;
 import java.util.Map;
 import org.apache.gravitino.catalog.hadoop.fs.FileSystemProvider;
+import org.apache.gravitino.catalog.hadoop.fs.FileSystemUtils;
+import org.apache.gravitino.storage.GCSProperties;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -30,15 +33,17 @@
 
 public class GCSFileSystemProvider implements FileSystemProvider {
   private static final Logger LOGGER = LoggerFactory.getLogger(GCSFileSystemProvider.class);
+  private static final String GCS_SERVICE_ACCOUNT_JSON_FILE =
+      "fs.gs.auth.service.account.json.keyfile";
+
+  private static final Map<String, String> GRAVITINO_KEY_TO_GCS_HADOOP_KEY =
+      ImmutableMap.of(GCSProperties.GCS_SERVICE_ACCOUNT_JSON_PATH, GCS_SERVICE_ACCOUNT_JSON_FILE);
 
   @Override
   public FileSystem getFileSystem(Path path, Map<String, String> config) throws IOException {
     Configuration configuration = new Configuration();
-    config.forEach(
-        (k, v) -> {
-          configuration.set(k.replace("gravitino.bypass.", ""), v);
-        });
-
+    FileSystemUtils.toHadoopConfigMap(config, GRAVITINO_KEY_TO_GCS_HADOOP_KEY)
+        .forEach(configuration::set);
     LOGGER.info("Creating GCS file system with config: {}", config);
     return GoogleHadoopFileSystem.newInstance(path.toUri(), configuration);
   }

catalogs/catalog-common/src/main/java/org/apache/gravitino/storage/GCSProperties.java

@@ -0,0 +1,28 @@
+/*
+ * 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.storage;
+
+public class GCSProperties {
+
+  // The path of service account JSON file of Google Cloud Storage.
+  public static final String GCS_SERVICE_ACCOUNT_JSON_PATH = "gcs-service-account-file";
+
+  private GCSProperties() {}
+}

catalogs/catalog-common/src/main/java/org/apache/gravitino/storage/OSSProperties.java

@@ -26,7 +26,7 @@ public class OSSProperties {
   // The static access key ID used to access OSS data.
   public static final String GRAVITINO_OSS_ACCESS_KEY_ID = "oss-access-key-id";
   // The static access key secret used to access OSS data.
-  public static final String GRAVITINO_OSS_ACCESS_KEY_SECRET = "oss-access-key-secret";
+  public static final String GRAVITINO_OSS_ACCESS_KEY_SECRET = "oss-secret-access-key";
 
   private OSSProperties() {}
 }

catalogs/catalog-common/src/main/java/org/apache/gravitino/storage/S3Properties.java

@@ -31,5 +31,8 @@ public class S3Properties {
   // The region of the S3 service.
   public static final String GRAVITINO_S3_REGION = "s3-region";
 
+  // The S3 credentials provider class name.
+  public static final String GRAVITINO_S3_CREDS_PROVIDER = "s3-creds-provider";
+
   private S3Properties() {}
 }

catalogs/catalog-hadoop/build.gradle.kts

@@ -36,6 +36,10 @@ dependencies {
     exclude(group = "*")
   }
 
+  implementation(project(":catalogs:catalog-common")) {
+    exclude(group = "*")
+  }
+
   compileOnly(libs.guava)
 
   implementation(libs.hadoop3.common) {

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

@@ -32,6 +32,20 @@
  */
 public interface FileSystemProvider {
 
+  /**
+   * The prefix of the configuration key that should be bypassed when setting the configuration to
+   * the FileSystem instance.
+   *
+   * <p>For example, if the configuration key passed to {@link
+   * FileSystemProvider#getFileSystem(Path, Map)} 'gravitino.bypass.fs.s3a.endpoint', the prefix
+   * 'gravitino.bypass.' should be removed when setting the configuration to the FileSystem
+   * instance.
+   *
+   * <p>User can use this prefix to pass the configuration item that has not been defined in
+   * Gravitino.
+   */
+  String GRAVITINO_BYPASS = "gravitino.bypass.";
+
   /**
    * Get the FileSystem instance according to the configuration map and file path.
    *

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

@@ -20,6 +20,7 @@
 
 import static org.apache.gravitino.catalog.hadoop.HadoopCatalogPropertiesMetadata.BUILTIN_HDFS_FS_PROVIDER;
 import static org.apache.gravitino.catalog.hadoop.HadoopCatalogPropertiesMetadata.BUILTIN_LOCAL_FS_PROVIDER;
+import static org.apache.gravitino.catalog.hadoop.fs.FileSystemProvider.GRAVITINO_BYPASS;
 
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
@@ -101,4 +102,45 @@ public static FileSystemProvider getFileSystemProviderByName(
                         "File system provider with name '%s' not found in the file system provider list.",
                         fileSystemProviderName)));
   }
+
+  /**
+   * Convert the Gravitino configuration to Hadoop configuration.
+   *
+   * <p>Predefined keys have the highest priority. If the key does not exist in the predefined keys,
+   * it will be set to the configuration. Keys with prefixes 'gravitino.bypass' has the lowest
+   * priority.
+   *
+   * @param config Gravitino configuration
+   * @return Hadoop configuration Map
+   */
+  public static Map<String, String> toHadoopConfigMap(
+      Map<String, String> config, Map<String, String> predefinedKeys) {
+    Map<String, String> result = Maps.newHashMap();
+    Set<String> highestPriorityKeys = Sets.newHashSet();
+    config.forEach(
+        (k, v) -> {
+          if (predefinedKeys.containsKey(k)) {
+            String key = predefinedKeys.get(k);
+            highestPriorityKeys.add(key);
+            result.put(key, v);
+          }
+
+          if (!k.startsWith(GRAVITINO_BYPASS)) {
+            // If the key does not start with 'gravitino.bypass' and is not in the highest priority
+            // keys, set the value to the configuration.
+            if (!highestPriorityKeys.contains(k)) {
+              result.put(k, v);
+            }
+          } else {
+            // If the key starts with 'gravitino.bypass', remove the prefix and set the value to the
+            // configuration if the key does not exist in the configuration.
+            String key = k.replace(GRAVITINO_BYPASS, "");
+            if (!result.containsKey(key)) {
+              result.put(key, v);
+            }
+          }
+        });
+
+    return result;
+  }
 }

catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/integration/test/HadoopGCSCatalogIT.java

@@ -19,19 +19,26 @@
 package org.apache.gravitino.catalog.hadoop.integration.test;
 
 import static org.apache.gravitino.catalog.hadoop.HadoopCatalogPropertiesMetadata.FILESYSTEM_PROVIDERS;
+import static org.apache.gravitino.storage.GCSProperties.GCS_SERVICE_ACCOUNT_JSON_PATH;
 
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import java.io.IOException;
 import java.net.URI;
 import java.util.Map;
 import org.apache.gravitino.Catalog;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Schema;
+import org.apache.gravitino.file.Fileset;
 import org.apache.gravitino.integration.test.util.GravitinoITUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
 
 @Tag("gravitino-docker-test")
 @Disabled(
@@ -95,10 +102,8 @@ protected String defaultBaseLocation() {
 
   protected void createCatalog() {
     Map<String, String> map = Maps.newHashMap();
-    map.put("gravitino.bypass.fs.gs.auth.service.account.enable", "true");
-    map.put("gravitino.bypass.fs.gs.auth.service.account.json.keyfile", SERVICE_ACCOUNT_FILE);
+    map.put(GCS_SERVICE_ACCOUNT_JSON_PATH, SERVICE_ACCOUNT_FILE);
     map.put(FILESYSTEM_PROVIDERS, "gcs");
-
     metalake.createCatalog(catalogName, Catalog.Type.FILESET, provider, "comment", map);
 
     catalog = metalake.loadCatalog(catalogName);
@@ -107,4 +112,64 @@ protected void createCatalog() {
   protected String generateLocation(String filesetName) {
     return String.format("%s/%s", defaultBaseLocation, filesetName);
   }
+
+  @Test
+  public void testCreateSchemaAndFilesetWithSpecialLocation() {
+    String localCatalogName = GravitinoITUtils.genRandomName("local_catalog");
+
+    String ossLocation = String.format("gs://%s", BUCKET_NAME);
+    Map<String, String> catalogProps = Maps.newHashMap();
+    catalogProps.put("location", ossLocation);
+    catalogProps.put(GCS_SERVICE_ACCOUNT_JSON_PATH, SERVICE_ACCOUNT_FILE);
+    catalogProps.put(FILESYSTEM_PROVIDERS, "gcs");
+
+    Catalog localCatalog =
+        metalake.createCatalog(
+            localCatalogName, Catalog.Type.FILESET, provider, "comment", catalogProps);
+    Assertions.assertEquals(ossLocation, localCatalog.properties().get("location"));
+
+    // Create schema without specifying location.
+    Schema localSchema =
+        localCatalog
+            .asSchemas()
+            .createSchema("local_schema", "comment", ImmutableMap.of("key1", "val1"));
+
+    Fileset localFileset =
+        localCatalog
+            .asFilesetCatalog()
+            .createFileset(
+                NameIdentifier.of(localSchema.name(), "local_fileset"),
+                "fileset comment",
+                Fileset.Type.MANAGED,
+                null,
+                ImmutableMap.of("k1", "v1"));
+    Assertions.assertEquals(
+        ossLocation + "/local_schema/local_fileset", localFileset.storageLocation());
+
+    // Delete schema
+    localCatalog.asSchemas().dropSchema(localSchema.name(), true);
+
+    // Create schema with specifying location.
+    Map<String, String> schemaProps = ImmutableMap.of("location", ossLocation);
+    Schema localSchema2 =
+        localCatalog.asSchemas().createSchema("local_schema2", "comment", schemaProps);
+    Assertions.assertEquals(ossLocation, localSchema2.properties().get("location"));
+
+    Fileset localFileset2 =
+        localCatalog
+            .asFilesetCatalog()
+            .createFileset(
+                NameIdentifier.of(localSchema2.name(), "local_fileset2"),
+                "fileset comment",
+                Fileset.Type.MANAGED,
+                null,
+                ImmutableMap.of("k1", "v1"));
+    Assertions.assertEquals(ossLocation + "/local_fileset2", localFileset2.storageLocation());
+
+    // Delete schema
+    localCatalog.asSchemas().dropSchema(localSchema2.name(), true);
+
+    // Delete catalog
+    metalake.dropCatalog(localCatalogName, true);
+  }
 }