[apache#4508]feat(iceberg-rest-server): support load custom catalog a…

…s backend in the Iceberg REST service (apache#4563)

### What changes were proposed in this pull request?

support load custom catalog as backend

### Why are the changes needed?

improvement: apache#4508

### Does this PR introduce _any_ user-facing change?

1. add a property key

### How was this patch tested?

1. add UT
2. manual test

---------

Co-authored-by: theoryxu <[email protected]>

catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java

@@ -21,6 +21,7 @@
 public class IcebergConstants {
   // Iceberg catalog properties constants
   public static final String CATALOG_BACKEND = "catalog-backend";
+  public static final String CATALOG_BACKEND_IMPL = "catalog-backend-impl";
 
   public static final String GRAVITINO_JDBC_USER = "jdbc-user";
   public static final String ICEBERG_JDBC_USER = "jdbc.user";

docs/iceberg-rest-service.md

@@ -162,6 +162,14 @@ If you have a JDBC Iceberg catalog prior, you must set `catalog-backend-name` to
 You must download the corresponding JDBC driver to the `iceberg-rest-server/libs` directory.
 :::
 
+#### Custom backend configuration
+| Configuration item                             | Description                                                                                                         | Default value    | Required | Since Version |
+|------------------------------------------------|---------------------------------------------------------------------------------------------------------------------|------------------|----------|---------------|
+| `gravitino.iceberg-rest.catalog-backend`       | The Catalog backend of the Gravitino Iceberg REST catalog service. Use the value **`custom`** for a Custom catalog. | `memory`         | Yes      | 0.2.0         |
+| `gravitino.iceberg-rest.catalog-backend-impl`  | The fully-qualified class name of a custom catalog implementation, only worked if `catalog-backend` is `custom`.    | (none)           | No       | 0.7.0         |
+
+If you want to use a custom Iceberg Catalog as `catalog-backend`, you can add a corresponding jar file to the classpath and load a custom Iceberg Catalog implementation by specifying the `catalog-backend-impl` property.
+
 #### Multi catalog support
 
 The Gravitino Iceberg REST server supports multiple catalogs and offers a configuration-based catalog management system.

iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/IcebergCatalogBackend.java

@@ -22,5 +22,6 @@ public enum IcebergCatalogBackend {
   HIVE,
   JDBC,
   MEMORY,
-  REST
+  REST,
+  CUSTOM
 }

iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/IcebergConfig.java

@@ -43,6 +43,14 @@ public class IcebergConfig extends Config implements OverwriteDefaultConfig {
           .stringConf()
           .createWithDefault("memory");
 
+  public static final ConfigEntry<String> CATALOG_BACKEND_IMPL =
+      new ConfigBuilder(IcebergConstants.CATALOG_BACKEND_IMPL)
+          .doc(
+              "The fully-qualified class name of a custom catalog implementation, only worked if `catalog-backend` is `custom`")
+          .version(ConfigConstants.VERSION_0_7_0)
+          .stringConf()
+          .create();
+
   public static final ConfigEntry<String> CATALOG_WAREHOUSE =
       new ConfigBuilder(IcebergConstants.WAREHOUSE)
           .doc("Warehouse directory of catalog")

iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java

@@ -39,6 +39,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdfs.HdfsConfiguration;
 import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.CatalogUtil;
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.hive.HiveCatalog;
 import org.apache.iceberg.inmemory.InMemoryCatalog;
@@ -149,6 +150,16 @@ private static Catalog loadRestCatalog(IcebergConfig icebergConfig) {
     return restCatalog;
   }
 
+  private static Catalog loadCustomCatalog(IcebergConfig icebergConfig) {
+    String customCatalogName = icebergConfig.getCatalogBackendName("custom");
+    String className = icebergConfig.get(IcebergConfig.CATALOG_BACKEND_IMPL);
+    return CatalogUtil.loadCatalog(
+        className,
+        customCatalogName,
+        icebergConfig.getIcebergCatalogProperties(),
+        new HdfsConfiguration());
+  }
+
   @VisibleForTesting
   static Catalog loadCatalogBackend(String catalogType) {
     return loadCatalogBackend(
@@ -168,6 +179,8 @@ public static Catalog loadCatalogBackend(
         return loadJdbcCatalog(icebergConfig);
       case REST:
         return loadRestCatalog(icebergConfig);
+      case CUSTOM:
+        return loadCustomCatalog(icebergConfig);
       default:
         throw new RuntimeException(
             catalogBackend

iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/CustomCatalogForTest.java

@@ -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.
+ */
+
+package org.apache.gravitino.iceberg.common.utils;
+
+import java.util.List;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+
+public class CustomCatalogForTest implements Catalog {
+  @Override
+  public List<TableIdentifier> listTables(Namespace namespace) {
+    return null;
+  }
+
+  @Override
+  public boolean dropTable(TableIdentifier identifier, boolean purge) {
+    return false;
+  }
+
+  @Override
+  public void renameTable(TableIdentifier from, TableIdentifier to) {}
+
+  @Override
+  public Table loadTable(TableIdentifier identifier) {
+    return null;
+  }
+}

iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java

@@ -78,4 +78,38 @@ void testLoadCatalog() {
           IcebergCatalogUtil.loadCatalogBackend("other");
         });
   }
+
+  @Test
+  void testValidLoadCustomCatalog() {
+    Catalog catalog;
+    Map<String, String> config = new HashMap<>();
+
+    config.put(
+        "catalog-backend-impl", "org.apache.gravitino.iceberg.common.utils.CustomCatalogForTest");
+    catalog =
+        IcebergCatalogUtil.loadCatalogBackend(
+            IcebergCatalogBackend.valueOf("CUSTOM"), new IcebergConfig(config));
+    Assertions.assertTrue(catalog instanceof CustomCatalogForTest);
+  }
+
+  @Test
+  void testInvalidLoadCustomCatalog() {
+    Assertions.assertThrowsExactly(
+        NullPointerException.class,
+        () ->
+            IcebergCatalogUtil.loadCatalogBackend(
+                IcebergCatalogBackend.valueOf("CUSTOM"), new IcebergConfig(new HashMap<>())));
+
+    Assertions.assertThrowsExactly(
+        IllegalArgumentException.class,
+        () ->
+            IcebergCatalogUtil.loadCatalogBackend(
+                IcebergCatalogBackend.valueOf("CUSTOM"),
+                new IcebergConfig(
+                    new HashMap<String, String>() {
+                      {
+                        put("catalog-backend-impl", "org.apache.");
+                      }
+                    })));
+  }
 }