[#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

build.gradle.kts

@@ -744,7 +744,8 @@ tasks {
       if (!it.name.startsWith("catalog") &&
         !it.name.startsWith("authorization") &&
         !it.name.startsWith("client") && !it.name.startsWith("filesystem") && !it.name.startsWith("spark") && !it.name.startsWith("iceberg") && it.name != "trino-connector" &&
-        it.name != "integration-test" && it.name != "hive-metastore-common" && !it.name.startsWith("flink")
+        it.name != "integration-test" && it.name != "hive-metastore-common" && !it.name.startsWith("flink") &&
+        it.name != "hadoop-gcs-bundle" && it.name != "hadoop-s3-bundle"
       ) {
         from(it.configurations.runtimeClasspath)
         into("distribution/package/libs")
@@ -763,7 +764,8 @@ tasks {
         !it.name.startsWith("integration-test") &&
         !it.name.startsWith("flink") &&
         !it.name.startsWith("trino-connector") &&
-        it.name != "hive-metastore-common"
+        it.name != "hive-metastore-common" &&
+        it.name != "hadoop-gcs-bundle" && it.name != "hadoop-s3-bundle"
       ) {
         dependsOn("${it.name}:build")
         from("${it.name}/build/libs")

bundles/build.gradle.kts

@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+tasks.all {
+    enabled = false
+}

bundles/hadoop-gcs-bundle/build.gradle.kts

@@ -0,0 +1,46 @@
+/*
+ * 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.
+ */
+import com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar
+
+plugins {
+  `maven-publish`
+  id("java")
+  alias(libs.plugins.shadow)
+}
+
+dependencies {
+  compileOnly(project(":catalogs:catalog-hadoop"))
+  compileOnly(libs.hadoop3.common)
+  implementation(libs.hadoop3.gcs)
+}
+
+tasks.withType(ShadowJar::class.java) {
+  isZip64 = true
+  configurations = listOf(project.configurations.runtimeClasspath.get())
+  archiveClassifier.set("")
+}
+
+tasks.jar {
+  dependsOn(tasks.named("shadowJar"))
+  archiveClassifier.set("empty")
+}
+
+tasks.compileJava {
+  dependsOn(":catalogs:catalog-hadoop:runtimeJars")
+}

bundles/hadoop-gcs-bundle/src/main/java/org/apache/gravitino/fileset/gcs/GCSConfigurationProvider.java

@@ -0,0 +1,31 @@
+/*
+ * 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.fileset.gcs;
+
+import java.util.Map;
+import org.apache.gravitino.catalog.hadoop.DefaultConfigurationProvider;
+import org.apache.hadoop.conf.Configuration;
+
+public class GCSConfigurationProvider extends DefaultConfigurationProvider {
+  // Add your own implementation here
+  @Override
+  public Configuration getConfiguration(Map<String, String> conf) {
+    return super.getConfiguration(conf);
+  }
+}

bundles/hadoop-s3-bundle/build.gradle.kts

@@ -0,0 +1,46 @@
+/*
+ * 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.
+ */
+import com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar
+
+plugins {
+  `maven-publish`
+  id("java")
+  alias(libs.plugins.shadow)
+}
+
+dependencies {
+  compileOnly(project(":catalogs:catalog-hadoop"))
+  compileOnly(libs.hadoop3.common)
+  implementation(libs.hadoop3.s3)
+}
+
+tasks.withType(ShadowJar::class.java) {
+  isZip64 = true
+  configurations = listOf(project.configurations.runtimeClasspath.get())
+  archiveClassifier.set("")
+}
+
+tasks.jar {
+  dependsOn(tasks.named("shadowJar"))
+  archiveClassifier.set("empty")
+}
+
+tasks.compileJava {
+  dependsOn(":catalogs:catalog-hadoop:runtimeJars")
+}

bundles/hadoop-s3-bundle/src/main/java/org/apache/gravitino/fileset/s3/S3ConfigurationProvider.java

@@ -0,0 +1,31 @@
+/*
+ * 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.fileset.s3;
+
+import java.util.Map;
+import org.apache.gravitino.catalog.hadoop.DefaultConfigurationProvider;
+import org.apache.hadoop.conf.Configuration;
+
+public class S3ConfigurationProvider extends DefaultConfigurationProvider {
+  // Add your own implementation here
+  @Override
+  public Configuration getConfiguration(Map<String, String> conf) {
+    return super.getConfiguration(conf);
+  }
+}

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

@@ -0,0 +1,37 @@
+/*
+ * 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;
+
+import java.util.Map;
+import org.apache.hadoop.conf.Configuration;
+
+public interface ConfigurationProvider {
+
+  default void initialize(Map<String, String> conf) {
+    // Do nothing;
+  }
+  /**
+   * Get the configuration from the given properties.
+   *
+   * @param conf The properties to get the configuration from.
+   * @return The configuration.
+   */
+  Configuration getConfiguration(Map<String, String> conf);
+}

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

@@ -0,0 +1,31 @@
+/*
+ * 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;
+
+import java.util.Map;
+import org.apache.hadoop.conf.Configuration;
+
+public class DefaultConfigurationProvider implements ConfigurationProvider {
+  @Override
+  public Configuration getConfiguration(Map<String, String> conf) {
+    Configuration configuration = new Configuration();
+    conf.forEach(configuration::set);
+    return configuration;
+  }
+}

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

@@ -132,6 +132,20 @@ public void initialize(
                     Map.Entry::getValue));
     bypassConfigs.forEach(hadoopConf::set);
 
+    String configProviderClass =
+        config.getOrDefault(
+            HadoopCatalogPropertiesMetadata.CONFIGURATION_PROVIDER,
+            DefaultConfigurationProvider.class.getCanonicalName());
+    try {
+      Class<?> providerClass = Class.forName(configProviderClass);
+      ConfigurationProvider provider =
+          (ConfigurationProvider) providerClass.getDeclaredConstructor().newInstance();
+      provider.initialize(bypassConfigs);
+      this.hadoopConf = provider.getConfiguration(bypassConfigs);
+    } catch (Exception e) {
+      throw new RuntimeException("Failed to initialize Hadoop configuration", e);
+    }
+
     String catalogLocation =
         (String)
             propertiesMetadata

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

@@ -34,6 +34,11 @@ 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 configuration provider class name, default value is {@link DefaultConfigurationProvider}.
+   */
+  public static final String CONFIGURATION_PROVIDER = "configuration.provider";
+
   private static final Map<String, PropertyEntry<?>> HADOOP_CATALOG_PROPERTY_ENTRIES =
       ImmutableMap.<String, PropertyEntry<?>>builder()
           .put(
@@ -44,6 +49,14 @@ public class HadoopCatalogPropertiesMetadata extends BaseCatalogPropertiesMetada
                   false /* immutable */,
                   null,
                   false /* hidden */))
+          .put(
+              CONFIGURATION_PROVIDER,
+              PropertyEntry.stringOptionalPropertyEntry(
+                  CONFIGURATION_PROVIDER,
+                  "The configuration provider class name",
+                  false /* immutable */,
+                  DefaultConfigurationProvider.class.getCanonicalName(),
+                  false /* hidden */))
           // The following two are about authentication.
           .putAll(KerberosConfig.KERBEROS_PROPERTY_ENTRIES)
           .putAll(AuthenticationConfig.AUTHENTICATION_PROPERTY_ENTRIES)

gradle/libs.versions.toml

@@ -32,6 +32,7 @@ airlift-resolver = "1.6"
 hive2 = "2.3.9"
 hadoop2 = "2.10.2"
 hadoop3 = "3.1.0"
+hadoop3-gcs = "3.0.0"
 hadoop-minikdc = "3.3.6"
 htrace-core4 = "4.1.0-incubating"
 httpclient5 = "5.2.1"
@@ -151,6 +152,8 @@ hadoop3-hdfs = { group = "org.apache.hadoop", name = "hadoop-hdfs", version.ref
 hadoop3-common = { group = "org.apache.hadoop", name = "hadoop-common", version.ref = "hadoop3"}
 hadoop3-client = { group = "org.apache.hadoop", name = "hadoop-client", version.ref = "hadoop3"}
 hadoop3-minicluster = { group = "org.apache.hadoop", name = "hadoop-minicluster", version.ref = "hadoop-minikdc"}
+hadoop3-s3 = { group = "org.apache.hadoop", name = "hadoop-aws", version.ref = "hadoop3"}
+hadoop3-gcs = { group = "com.google.cloud.bigdataoss", name = "gcs-connector", version.ref = "hadoop3-gcs"}
 htrace-core4 = { group = "org.apache.htrace", name = "htrace-core4", version.ref = "htrace-core4" }
 airlift-json = { group = "io.airlift", name = "json", version.ref = "airlift-json"}
 airlift-resolver = { group = "io.airlift.resolver", name = "resolver", version.ref = "airlift-resolver"}

settings.gradle.kts

@@ -70,3 +70,6 @@ project(":spark-connector:spark-runtime-3.5").projectDir = file("spark-connector
 include("web:web", "web:integration-test")
 include("docs")
 include("integration-test-common")
+include("bundles:hadoop-s3-bundle")
+include("bundles:hadoop-gcs-bundle")
+findProject(":bundles:hadoop-gcs-bundle")?.name = "hadoop-gcs-bundle"