[#6035]feat(spark-connector):Support custom catalog backend (#6036)

### What changes were proposed in this pull request?
When the catalog backend is `custom`, add the `catalog-impl` instead
`type` in properties

### Why are the changes needed?

Fix: #6035 

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

### How was this patch tested?
add a new case in `TestIcebergPropertiesConverter`

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

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.gravitino.iceberg.common;
+package org.apache.gravitino.catalog.lakehouse.iceberg;
 
 public enum IcebergCatalogBackend {
   HIVE,

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

@@ -37,6 +37,7 @@ public class IcebergPropertiesUtils {
   static {
     Map<String, String> map = new HashMap();
     map.put(IcebergConstants.CATALOG_BACKEND, IcebergConstants.CATALOG_BACKEND);
+    map.put(IcebergConstants.CATALOG_BACKEND_IMPL, IcebergConstants.CATALOG_BACKEND_IMPL);
     map.put(IcebergConstants.GRAVITINO_JDBC_DRIVER, IcebergConstants.GRAVITINO_JDBC_DRIVER);
     map.put(IcebergConstants.GRAVITINO_JDBC_USER, IcebergConstants.ICEBERG_JDBC_USER);
     map.put(IcebergConstants.GRAVITINO_JDBC_PASSWORD, IcebergConstants.ICEBERG_JDBC_PASSWORD);

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

@@ -30,7 +30,6 @@
 import java.util.Map;
 import org.apache.gravitino.connector.BaseCatalogPropertiesMetadata;
 import org.apache.gravitino.connector.PropertyEntry;
-import org.apache.gravitino.iceberg.common.IcebergCatalogBackend;
 import org.apache.gravitino.iceberg.common.authentication.AuthenticationConfig;
 import org.apache.gravitino.iceberg.common.authentication.kerberos.KerberosConfig;
 import org.apache.gravitino.storage.AzureProperties;

catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/integration/test/CatalogIcebergBaseIT.java

@@ -47,14 +47,14 @@
 import org.apache.gravitino.SchemaChange;
 import org.apache.gravitino.SupportsSchemas;
 import org.apache.gravitino.auth.AuthConstants;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
 import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergSchemaPropertiesMetadata;
 import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergTable;
 import org.apache.gravitino.catalog.lakehouse.iceberg.ops.IcebergCatalogWrapperHelper;
 import org.apache.gravitino.client.GravitinoMetalake;
 import org.apache.gravitino.exceptions.NoSuchSchemaException;
 import org.apache.gravitino.exceptions.SchemaAlreadyExistsException;
 import org.apache.gravitino.exceptions.TableAlreadyExistsException;
-import org.apache.gravitino.iceberg.common.IcebergCatalogBackend;
 import org.apache.gravitino.iceberg.common.IcebergConfig;
 import org.apache.gravitino.iceberg.common.utils.IcebergCatalogUtil;
 import org.apache.gravitino.integration.test.container.ContainerSuite;

docs/spark-connector/spark-catalog-iceberg.md

@@ -103,7 +103,8 @@ Gravitino spark connector will transform below property names which are defined
 
 | Gravitino catalog property name | Spark Iceberg connector configuration | Description                                                                                                                                                                                                         | Since Version    |
 |---------------------------------|---------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|------------------|
-| `catalog-backend`               | `type`                                | Catalog backend type                                                                                                                                                                                                | 0.5.0            |
+| `catalog-backend`               | `type`                                | Catalog backend type.Supports `hive` or `jdbc` or `rest` or `custom`                                                                                                                                                | 0.5.0            |
+| `catalog-backend-impl`          | `catalog-impl`                        | The fully-qualified class name of a custom catalog implementation, only worked if `catalog-backend` is `custom`                                                                                                     | 0.8.0-incubating |
 | `uri`                           | `uri`                                 | Catalog backend uri                                                                                                                                                                                                 | 0.5.0            |
 | `warehouse`                     | `warehouse`                           | Catalog backend warehouse                                                                                                                                                                                           | 0.5.0            |
 | `jdbc-user`                     | `jdbc.user`                           | JDBC user name                                                                                                                                                                                                      | 0.5.0            |

iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java

@@ -28,7 +28,7 @@
 import lombok.Getter;
 import lombok.Setter;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.gravitino.iceberg.common.IcebergCatalogBackend;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
 import org.apache.gravitino.iceberg.common.IcebergConfig;
 import org.apache.gravitino.iceberg.common.utils.IcebergCatalogUtil;
 import org.apache.gravitino.utils.IsolatedClassLoader;

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

@@ -30,10 +30,10 @@
 import java.util.HashMap;
 import java.util.Locale;
 import java.util.Map;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
 import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
 import org.apache.gravitino.exceptions.ConnectionFailedException;
 import org.apache.gravitino.iceberg.common.ClosableHiveCatalog;
-import org.apache.gravitino.iceberg.common.IcebergCatalogBackend;
 import org.apache.gravitino.iceberg.common.IcebergConfig;
 import org.apache.gravitino.iceberg.common.authentication.AuthenticationConfig;
 import org.apache.gravitino.iceberg.common.authentication.kerberos.HiveBackendProxy;

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

@@ -21,8 +21,8 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
 import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
-import org.apache.gravitino.iceberg.common.IcebergCatalogBackend;
 import org.apache.gravitino.iceberg.common.IcebergConfig;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.catalog.Catalog;

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTHiveCatalogIT.java

@@ -20,7 +20,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
-import org.apache.gravitino.iceberg.common.IcebergCatalogBackend;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
 import org.apache.gravitino.iceberg.common.IcebergConfig;
 import org.apache.gravitino.integration.test.container.ContainerSuite;
 import org.apache.gravitino.integration.test.container.HiveContainer;

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTJdbcCatalogIT.java

@@ -20,8 +20,8 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
 import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
-import org.apache.gravitino.iceberg.common.IcebergCatalogBackend;
 import org.apache.gravitino.iceberg.common.IcebergConfig;
 import org.apache.gravitino.integration.test.container.ContainerSuite;
 import org.apache.gravitino.integration.test.container.HiveContainer;

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTMemoryCatalogIT.java

@@ -20,7 +20,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
-import org.apache.gravitino.iceberg.common.IcebergCatalogBackend;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
 import org.apache.gravitino.iceberg.common.IcebergConfig;
 
 public class IcebergRESTMemoryCatalogIT extends IcebergRESTServiceIT {

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTServiceBaseIT.java

@@ -31,7 +31,7 @@
 import java.util.stream.IntStream;
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.gravitino.iceberg.common.IcebergCatalogBackend;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
 import org.apache.gravitino.iceberg.common.IcebergConfig;
 import org.apache.gravitino.iceberg.integration.test.util.IcebergRESTServerManager;
 import org.apache.gravitino.integration.test.util.ITUtils;

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRestKerberosHiveCatalogIT.java

@@ -25,7 +25,7 @@
 import java.util.Map;
 import java.util.Objects;
 import org.apache.commons.io.FileUtils;
-import org.apache.gravitino.iceberg.common.IcebergCatalogBackend;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
 import org.apache.gravitino.iceberg.common.IcebergConfig;
 import org.apache.gravitino.integration.test.container.HiveContainer;
 import org.apache.gravitino.integration.test.util.GravitinoITUtils;

spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergPropertiesConstants.java

@@ -29,6 +29,7 @@ public class IcebergPropertiesConstants {
   public static final String GRAVITINO_ICEBERG_CATALOG_BACKEND = IcebergConstants.CATALOG_BACKEND;
 
   static final String ICEBERG_CATALOG_TYPE = CatalogUtil.ICEBERG_CATALOG_TYPE;
+  static final String ICEBERG_CATALOG_IMPL = CatalogProperties.CATALOG_IMPL;
 
   public static final String GRAVITINO_ICEBERG_CATALOG_WAREHOUSE = IcebergConstants.WAREHOUSE;
 

spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergPropertiesConverter.java

@@ -23,6 +23,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
 import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
 import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergPropertiesUtils;
 import org.apache.gravitino.spark.connector.PropertiesConverter;
@@ -49,7 +50,20 @@ public Map<String, String> toSparkCatalogProperties(Map<String, String> properti
     Preconditions.checkArgument(
         StringUtils.isNotBlank(catalogBackend),
         String.format("%s should not be empty", IcebergConstants.CATALOG_BACKEND));
-    all.put(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE, catalogBackend);
+    if (catalogBackend.equalsIgnoreCase(IcebergCatalogBackend.CUSTOM.name())) {
+      String catalogBackendImpl = all.remove(IcebergConstants.CATALOG_BACKEND_IMPL);
+      Preconditions.checkArgument(
+          StringUtils.isNotBlank(catalogBackendImpl),
+          String.format(
+              "%s should not be empty when %s is %s",
+              IcebergConstants.CATALOG_BACKEND_IMPL,
+              IcebergConstants.CATALOG_BACKEND,
+              IcebergCatalogBackend.CUSTOM.name()));
+      all.put(IcebergPropertiesConstants.ICEBERG_CATALOG_IMPL, catalogBackendImpl);
+    } else {
+      all.put(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE, catalogBackend);
+    }
+
     all.put(IcebergPropertiesConstants.ICEBERG_CATALOG_CACHE_ENABLED, "FALSE");
     return all;
   }

spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestIcebergPropertiesConverter.java

@@ -21,6 +21,8 @@
 
 import com.google.common.collect.ImmutableMap;
 import java.util.Map;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -113,4 +115,29 @@ void testCatalogPropertiesWithRestBackend() {
             "rest-warehouse"),
         properties);
   }
+
+  @Test
+  void testCatalogPropertiesWithCustomBackend() {
+    Map<String, String> properties =
+        icebergPropertiesConverter.toSparkCatalogProperties(
+            ImmutableMap.of(
+                IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND,
+                IcebergCatalogBackend.CUSTOM.name(),
+                IcebergConstants.CATALOG_BACKEND_IMPL,
+                "CustomCatalog",
+                IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE,
+                "custom-warehouse",
+                "key1",
+                "value1"));
+
+    Assertions.assertEquals(
+        ImmutableMap.of(
+            IcebergPropertiesConstants.ICEBERG_CATALOG_CACHE_ENABLED,
+            "FALSE",
+            IcebergPropertiesConstants.ICEBERG_CATALOG_IMPL,
+            "CustomCatalog",
+            IcebergPropertiesConstants.ICEBERG_CATALOG_WAREHOUSE,
+            "custom-warehouse"),
+        properties);
+  }
 }